Merge pull request #216 from trezor/tsusanka/stellar

Stellar implementation
pull/25/head
Jan Pochyla 6 years ago committed by GitHub
commit 7c36fc49f4
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23

@ -32,4 +32,4 @@ known_standard_library = micropython,ubinascii,ustruct,uctypes,utime,utimeq,trez
[tool:pytest]
addopts = --pyargs trezorlib.tests.device_tests
xfail_strict = true
run_xfail =
run_xfail = stellar

@ -43,3 +43,8 @@ def derive_node_without_passphrase(
node = bip32.from_seed(seed, curve_name)
node.derive_path(path)
return node
def remove_ed25519_prefix(pubkey: bytes) -> bytes:
# 0x01 prefix is not part of the actual public key, hence removed
return pubkey[1:]

@ -18,7 +18,6 @@ async def get_address(ctx, msg):
address = node.nem_address(network)
if msg.show_display:
while True:
if await _show_address(ctx, address, network):
break

@ -18,7 +18,7 @@ async def sign_tx(ctx, msg: NEMSignTx):
await multisig.ask(ctx, msg)
common = msg.multisig
else:
public_key = _get_public_key(node)
public_key = seed.remove_ed25519_prefix(node.public_key())
common = msg.transaction
if msg.transfer:
@ -48,10 +48,15 @@ async def sign_tx(ctx, msg: NEMSignTx):
# wrap transaction in multisig wrapper
if msg.cosigning:
tx = multisig.cosign(
_get_public_key(node), msg.transaction, tx, msg.multisig.signer
seed.remove_ed25519_prefix(node.public_key()),
msg.transaction,
tx,
msg.multisig.signer,
)
else:
tx = multisig.initiate(_get_public_key(node), msg.transaction, tx)
tx = multisig.initiate(
seed.remove_ed25519_prefix(node.public_key()), msg.transaction, tx
)
signature = ed25519.sign(node.private_key(), tx, NEM_HASH_ALG)
@ -59,8 +64,3 @@ async def sign_tx(ctx, msg: NEMSignTx):
resp.data = tx
resp.signature = signature
return resp
def _get_public_key(node) -> bytes:
# 0x01 prefix is not part of the actual public key, hence removed
return node.public_key()[1:]

@ -0,0 +1,33 @@
# Stellar
MAINTAINER = Tomas Susanka <tomas.susanka@satoshilabs.com>
AUTHOR = Tomas Susanka <tomas.susanka@satoshilabs.com>
REVIEWER = Jan Pochyla <jan.pochyla@satoshilabs.com>
ADVISORS = ZuluCrypto
-----
This implementation of Stellar for Trezor Core is mostly based on the trezor-mcu C implementation by ZuluCrypto.
Stellar has a comprehensive [developers documentation](https://www.stellar.org/developers/).
## Transactions
Stellar transaction is composed of one or more operations. We support all [operations](https://www.stellar.org/developers/guides/concepts/list-of-operations.html) except the Inflation operation (see [here](https://github.com/trezor/trezor-core/issues/202#issuecomment-392729595) for rationale). A list of supported operations:
- Account Merge
- Allow Trust
- Bump Sequence
- Change Trust
- Create Account
- Create Passive Offer
- Manage Data
- Manage Offer
- Path Payment
- Payment
- Set Options
Since a simple transaction can be composed of several operations, first the StellarSignTx message is sent to Trezor, which includes the total number of operations. Then the different operations are consecutively send to Trezor.

@ -0,0 +1,30 @@
from trezor.messages.MessageType import (
StellarGetAddress,
StellarGetPublicKey,
StellarSignTx,
)
from trezor.wire import protobuf_workflow, register
def dispatch_StellarGetAddress(*args, **kwargs):
from .get_address import get_address
return get_address(*args, **kwargs)
def dispatch_StellarGetPublicKey(*args, **kwargs):
from .get_public_key import get_public_key
return get_public_key(*args, **kwargs)
def dispatch_StellarSignTx(*args, **kwargs):
from .sign_tx import sign_tx
return sign_tx(*args, **kwargs)
def boot():
register(StellarGetAddress, protobuf_workflow, dispatch_StellarGetAddress)
register(StellarGetPublicKey, protobuf_workflow, dispatch_StellarGetPublicKey)
register(StellarSignTx, protobuf_workflow, dispatch_StellarSignTx)

@ -0,0 +1,73 @@
from micropython import const
from trezor.messages import MessageType
STELLAR_CURVE = "ed25519"
TX_TYPE = bytearray("\x00\x00\x00\x02")
# source: https://github.com/stellar/go/blob/3d2c1defe73dbfed00146ebe0e8d7e07ce4bb1b6/xdr/Stellar-transaction.x#L16
# Inflation not supported see https://github.com/trezor/trezor-core/issues/202#issuecomment-393342089
op_codes = {
"StellarAccountMergeOp": const(8),
"StellarAllowTrustOp": const(7),
"StellarBumpSequenceOp": const(11),
"StellarChangeTrustOp": const(6),
"StellarCreateAccountOp": const(0),
"StellarCreatePassiveOfferOp": const(4),
"StellarManageDataOp": const(10),
"StellarManageOfferOp": const(3),
"StellarPathPaymentOp": const(2),
"StellarPaymentOp": const(1),
"StellarSetOptionsOp": const(5),
}
op_wire_types = [
MessageType.StellarAccountMergeOp,
MessageType.StellarAllowTrustOp,
MessageType.StellarBumpSequenceOp,
MessageType.StellarChangeTrustOp,
MessageType.StellarCreateAccountOp,
MessageType.StellarCreatePassiveOfferOp,
MessageType.StellarManageDataOp,
MessageType.StellarManageOfferOp,
MessageType.StellarPathPaymentOp,
MessageType.StellarPaymentOp,
MessageType.StellarSetOptionsOp,
]
# https://github.com/stellar/go/blob/e0ffe19f58879d3c31e2976b97a5bf10e13a337b/xdr/xdr_generated.go#L584
ASSET_TYPE_NATIVE = const(0)
ASSET_TYPE_ALPHANUM4 = const(1)
ASSET_TYPE_ALPHANUM12 = const(2)
# https://www.stellar.org/developers/guides/concepts/accounts.html#balance
# https://github.com/stellar/go/blob/3d2c1defe73dbfed00146ebe0e8d7e07ce4bb1b6/amount/main.go#L23
AMOUNT_DIVISIBILITY = const(7)
# https://github.com/stellar/go/blob/master/network/main.go
NETWORK_PASSPHRASE_PUBLIC = "Public Global Stellar Network ; September 2015"
NETWORK_PASSPHRASE_TESTNET = "Test SDF Network ; September 2015"
# https://www.stellar.org/developers/guides/concepts/accounts.html#flags
FLAG_AUTH_REQUIRED = const(1)
FLAG_AUTH_REVOCABLE = const(2)
FLAG_AUTH_IMMUTABLE = const(4)
FLAGS_MAX_SIZE = const(7)
# https://github.com/stellar/go/blob/e0ffe19f58879d3c31e2976b97a5bf10e13a337b/xdr/Stellar-transaction.x#L275
MEMO_TYPE_NONE = const(0)
MEMO_TYPE_TEXT = const(1)
MEMO_TYPE_ID = const(2)
MEMO_TYPE_HASH = const(3)
MEMO_TYPE_RETURN = const(4)
# https://github.com/stellar/go/blob/3d2c1defe73dbfed00146ebe0e8d7e07ce4bb1b6/xdr/xdr_generated.go#L156
SIGN_TYPE_ACCOUNT = const(0)
SIGN_TYPE_PRE_AUTH = const(1)
SIGN_TYPE_HASH = const(2)
def get_op_code(msg) -> int:
if msg.__qualname__ not in op_codes:
raise ValueError("Stellar: op code unknown")
return op_codes[msg.__qualname__]

@ -0,0 +1,21 @@
from trezor.messages.StellarAddress import StellarAddress
from trezor.messages.StellarGetAddress import StellarGetAddress
from apps.common import seed
from apps.common.display_address import show_address, show_qr
from apps.stellar import helpers
async def get_address(ctx, msg: StellarGetAddress):
node = await seed.derive_node(ctx, msg.address_n, helpers.STELLAR_CURVE)
pubkey = seed.remove_ed25519_prefix(node.public_key())
address = helpers.address_from_public_key(pubkey)
if msg.show_display:
while True:
if await show_address(ctx, address):
break
if await show_qr(ctx, address.upper()):
break
return StellarAddress(address=address)

@ -0,0 +1,34 @@
from ubinascii import hexlify
from trezor import ui
from trezor.messages import ButtonRequestType
from trezor.messages.StellarGetPublicKey import StellarGetPublicKey
from trezor.messages.StellarPublicKey import StellarPublicKey
from trezor.ui.text import Text
from apps.common import seed
from apps.common.confirm import confirm
from apps.common.display_address import split_address
from apps.stellar import helpers
async def get_public_key(ctx, msg: StellarGetPublicKey):
node = await seed.derive_node(ctx, msg.address_n, helpers.STELLAR_CURVE)
pubkey = seed.remove_ed25519_prefix(node.public_key())
if msg.show_display:
while True:
if await _show(ctx, pubkey):
break
return StellarPublicKey(public_key=pubkey)
async def _show(ctx, pubkey: bytes):
lines = split_address(hexlify(pubkey))
text = Text("Export Stellar ID", ui.ICON_RECEIVE, icon_color=ui.GREEN)
text.mono(*lines)
return await confirm(
ctx, text, code=ButtonRequestType.Address, cancel_style=ui.BTN_KEY
)

@ -0,0 +1,52 @@
import ustruct
from trezor.crypto import base32
from trezor.wire import ProcessError
STELLAR_CURVE = "ed25519"
def public_key_from_address(address: str) -> bytes:
"""Extracts public key from an address
Stellar address is in format:
<1-byte version> <32-bytes ed25519 public key> <2-bytes CRC-16 checksum>
"""
b = base32.decode(address)
_crc16_checksum_verify(b[:-2], b[-2:])
return b[1:-2]
def address_from_public_key(pubkey: bytes):
"""Returns the base32-encoded version of public key bytes (G...)"""
address = bytearray()
address.append(6 << 3) # version -> 'G'
address.extend(pubkey)
address.extend(_crc16_checksum(bytes(address))) # checksum
return base32.encode(address)
def _crc16_checksum_verify(data: bytes, checksum: bytes):
if _crc16_checksum(data) != checksum:
raise ProcessError("Invalid address checksum")
def _crc16_checksum(data: bytes) -> bytes:
"""Returns the CRC-16 checksum of bytearray bytes
Ported from Java implementation at: http://introcs.cs.princeton.edu/java/61data/CRC16CCITT.java.html
Initial value changed to 0x0000 to match Stellar configuration.
"""
crc = 0x0000
polynomial = 0x1021
for byte in data:
for i in range(8):
bit = (byte >> (7 - i) & 1) == 1
c15 = (crc >> 15 & 1) == 1
crc <<= 1
if c15 ^ bit:
crc ^= polynomial
return ustruct.pack("<H", crc & 0xffff)

@ -0,0 +1,81 @@
from trezor import ui, utils
from trezor.messages import ButtonRequestType
from trezor.ui.text import Text
from apps.common.confirm import require_confirm, require_hold_to_confirm
from apps.stellar import consts
async def require_confirm_init(ctx, address: str, network_passphrase: str):
text = Text("Confirm Stellar", ui.ICON_SEND, icon_color=ui.GREEN)
text.normal("Initialize singing with")
text.mono(*split(address))
await require_confirm(ctx, text, ButtonRequestType.ConfirmOutput)
network = get_network_warning(network_passphrase)
if network:
text = Text("Confirm network", ui.ICON_CONFIRM, icon_color=ui.GREEN)
text.normal("Transaction is on")
text.bold(network)
await require_confirm(ctx, text, ButtonRequestType.ConfirmOutput)
async def require_confirm_memo(ctx, memo_type: int, memo_text: str):
text = Text("Confirm memo", ui.ICON_CONFIRM, icon_color=ui.GREEN)
if memo_type == consts.MEMO_TYPE_TEXT:
text.bold("Memo (TEXT)")
elif memo_type == consts.MEMO_TYPE_ID:
text.bold("Memo (ID)")
elif memo_type == consts.MEMO_TYPE_HASH:
text.bold("Memo (HASH)")
elif memo_type == consts.MEMO_TYPE_RETURN:
text.bold("Memo (RETURN)")
else: # MEMO_TYPE_NONE
text.bold("No memo set!")
text.normal("Important: Many exchanges require a memo when depositing")
if memo_type != consts.MEMO_TYPE_NONE:
text.mono(*split(memo_text))
await require_confirm(ctx, text, ButtonRequestType.ConfirmOutput)
async def require_confirm_final(ctx, fee: int, num_operations: int):
op_str = str(num_operations) + " operation"
if num_operations > 1:
op_str += "s"
text = Text("Final confirm", ui.ICON_SEND, icon_color=ui.GREEN)
text.normal("Sign this transaction")
text.normal("made up of " + op_str)
text.bold("and pay " + format_amount(fee))
text.normal("for fee?")
# we use SignTx, not ConfirmOutput, for compatibility with T1
await require_hold_to_confirm(ctx, text, ButtonRequestType.SignTx)
def format_amount(amount: int, ticker=True) -> str:
t = ""
if ticker:
t = " XLM"
return utils.format_amount(amount, consts.AMOUNT_DIVISIBILITY) + t
def split(text):
return utils.chunks(text, 17)
def trim(payload: str, length: int, dots=True) -> str:
if len(payload) > length:
if dots:
return payload[: length - 2] + ".."
return payload[: length - 2]
return payload
def trim_to_rows(payload: str, rows: int = 1) -> str:
return trim(payload, rows * 17)
def get_network_warning(network_passphrase: str):
if network_passphrase == consts.NETWORK_PASSPHRASE_PUBLIC:
return None
if network_passphrase == consts.NETWORK_PASSPHRASE_TESTNET:
return "testnet network"
return "private network"

@ -0,0 +1,44 @@
from apps.stellar import consts, writers
from apps.stellar.operations import layout, serialize
async def operation(ctx, w, op):
if op.source_account:
await layout.confirm_source_account(ctx, op.source_account)
serialize.serialize_account(w, op.source_account)
writers.write_uint32(w, consts.get_op_code(op))
if isinstance(op, serialize.StellarAccountMergeOp):
await layout.confirm_account_merge_op(ctx, op)
serialize.serialize_account_merge_op(w, op)
elif isinstance(op, serialize.StellarAllowTrustOp):
await layout.confirm_allow_trust_op(ctx, op)
serialize.serialize_allow_trust_op(w, op)
elif isinstance(op, serialize.StellarBumpSequenceOp):
await layout.confirm_bump_sequence_op(ctx, op)
serialize.serialize_bump_sequence_op(w, op)
elif isinstance(op, serialize.StellarChangeTrustOp):
await layout.confirm_change_trust_op(ctx, op)
serialize.serialize_change_trust_op(w, op)
elif isinstance(op, serialize.StellarCreateAccountOp):
await layout.confirm_create_account_op(ctx, op)
serialize.serialize_create_account_op(w, op)
elif isinstance(op, serialize.StellarCreatePassiveOfferOp):
await layout.confirm_create_passive_offer_op(ctx, op)
serialize.serialize_create_passive_offer_op(w, op)
elif isinstance(op, serialize.StellarManageDataOp):
await layout.confirm_manage_data_op(ctx, op)
serialize.serialize_manage_data_op(w, op)
elif isinstance(op, serialize.StellarManageOfferOp):
await layout.confirm_manage_offer_op(ctx, op)
serialize.serialize_manage_offer_op(w, op)
elif isinstance(op, serialize.StellarPathPaymentOp):
await layout.confirm_path_payment_op(ctx, op)
serialize.serialize_path_payment_op(w, op)
elif isinstance(op, serialize.StellarPaymentOp):
await layout.confirm_payment_op(ctx, op)
serialize.serialize_payment_op(w, op)
elif isinstance(op, serialize.StellarSetOptionsOp):
await layout.confirm_set_options_op(ctx, op)
serialize.serialize_set_options_op(w, op)
else:
raise ValueError("serialize.Stellar: unknown operation")

@ -0,0 +1,250 @@
from ubinascii import hexlify
from trezor.messages import ButtonRequestType
from trezor.messages.StellarAccountMergeOp import StellarAccountMergeOp
from trezor.messages.StellarAllowTrustOp import StellarAllowTrustOp
from trezor.messages.StellarAssetType import StellarAssetType
from trezor.messages.StellarBumpSequenceOp import StellarBumpSequenceOp
from trezor.messages.StellarChangeTrustOp import StellarChangeTrustOp
from trezor.messages.StellarCreateAccountOp import StellarCreateAccountOp
from trezor.messages.StellarCreatePassiveOfferOp import StellarCreatePassiveOfferOp
from trezor.messages.StellarManageDataOp import StellarManageDataOp
from trezor.messages.StellarManageOfferOp import StellarManageOfferOp
from trezor.messages.StellarPathPaymentOp import StellarPathPaymentOp
from trezor.messages.StellarPaymentOp import StellarPaymentOp
from trezor.messages.StellarSetOptionsOp import StellarSetOptionsOp
from trezor.ui.text import Text
from trezor.wire import ProcessError
from apps.stellar import consts
from apps.stellar.layout import format_amount, require_confirm, split, trim_to_rows, ui
async def confirm_source_account(ctx, source_account: bytes):
text = Text("Confirm operation", ui.ICON_CONFIRM, icon_color=ui.GREEN)
text.bold("Source account:")
text.mono(*split(source_account))
await require_confirm(ctx, text, ButtonRequestType.ConfirmOutput)
async def confirm_allow_trust_op(ctx, op: StellarAllowTrustOp):
if op.is_authorized:
t = "Allow Trust"
else:
t = "Revoke Trust"
text = Text("Confirm operation", ui.ICON_CONFIRM, icon_color=ui.GREEN)
text.bold(t)
text.normal("of %s by:" % op.asset_code)
text.mono(*split(trim_to_rows(op.trusted_account, 3)))
await require_confirm(ctx, text, ButtonRequestType.ConfirmOutput)
async def confirm_account_merge_op(ctx, op: StellarAccountMergeOp):
text = Text("Confirm operation", ui.ICON_CONFIRM, icon_color=ui.GREEN)
text.bold("Account Merge")
text.normal("All XLM will be sent to:")
text.mono(*split(trim_to_rows(op.destination_account, 3)))
await require_confirm(ctx, text, ButtonRequestType.ConfirmOutput)
async def confirm_bump_sequence_op(ctx, op: StellarBumpSequenceOp):
text = Text("Confirm operation", ui.ICON_CONFIRM, icon_color=ui.GREEN)
text.bold("Bump Sequence")
text.normal("Set sequence to")
text.mono(str(op.bump_to))
await require_confirm(ctx, text, ButtonRequestType.ConfirmOutput)
async def confirm_change_trust_op(ctx, op: StellarChangeTrustOp):
if op.limit == 0:
t = "Delete Trust"
else:
t = "Add Trust"
text = Text("Confirm operation", ui.ICON_CONFIRM, icon_color=ui.GREEN)
text.bold(t)
text.normal("Asset: %s" % op.asset.code)
text.normal("Amount: %s" % format_amount(op.limit, ticker=False))
await require_confirm(ctx, text, ButtonRequestType.ConfirmOutput)
await confirm_asset_issuer(ctx, op.asset)
async def confirm_create_account_op(ctx, op: StellarCreateAccountOp):
text = Text("Confirm operation", ui.ICON_CONFIRM, icon_color=ui.GREEN)
text.bold("Create Account")
text.normal("with %s" % format_amount(op.starting_balance))
text.mono(*split(trim_to_rows(op.new_account, 3)))
await require_confirm(ctx, text, ButtonRequestType.ConfirmOutput)
async def confirm_create_passive_offer_op(ctx, op: StellarCreatePassiveOfferOp):
if op.amount == 0:
text = "Delete Passive Offer"
else:
text = "New Passive Offer"
await _confirm_offer(ctx, text, op)
async def confirm_manage_offer_op(ctx, op: StellarManageOfferOp):
if op.offer_id == 0:
text = "New Offer"
else:
if op.amount == 0:
text = "Delete"
else:
text = "Update"
text += " #%d" % op.offer_id
await _confirm_offer(ctx, text, op)
async def _confirm_offer(ctx, title, op):
text = Text("Confirm operation", ui.ICON_CONFIRM, icon_color=ui.GREEN)
text.bold(title)
text.normal(
"Sell %s %s" % (format_amount(op.amount, ticker=False), op.selling_asset.code)
)
text.normal("For %f" % (op.price_n / op.price_d))
text.normal("Per %s" % format_asset_code(op.buying_asset))
await require_confirm(ctx, text, ButtonRequestType.ConfirmOutput)
await confirm_asset_issuer(ctx, op.selling_asset)
await confirm_asset_issuer(ctx, op.buying_asset)
async def confirm_manage_data_op(ctx, op: StellarManageDataOp):
from trezor.crypto.hashlib import sha256
if op.value:
title = "Set"
else:
title = "Clear"
text = Text("Confirm operation", ui.ICON_CONFIRM, icon_color=ui.GREEN)
text.bold("%s data value key" % title)
text.mono(*split(op.key))
await require_confirm(ctx, text, ButtonRequestType.ConfirmOutput)
if op.value:
digest = sha256(op.value).digest()
digest_str = hexlify(digest).decode()
text = Text("Confirm operation", ui.ICON_CONFIRM, icon_color=ui.GREEN)
text.bold("Value (SHA-256):")
text.mono(*split(digest_str))
await require_confirm(ctx, text, ButtonRequestType.ConfirmOutput)
async def confirm_path_payment_op(ctx, op: StellarPathPaymentOp):
text = Text("Confirm operation", ui.ICON_CONFIRM, icon_color=ui.GREEN)
text.bold("Path Pay %s" % format_amount(op.destination_amount, ticker=False))
text.bold("%s to:" % format_asset_code(op.destination_asset))
text.mono(*split(trim_to_rows(op.destination_account, 3)))
await require_confirm(ctx, text, ButtonRequestType.ConfirmOutput)
await confirm_asset_issuer(ctx, op.destination_asset)
# confirm what the sender is using to pay
text = Text("Confirm operation", ui.ICON_CONFIRM, icon_color=ui.GREEN)
text.normal("Pay using")
text.bold(format_amount(op.send_max, ticker=False))
text.bold(format_asset_code(op.send_asset))
text.normal("This amount is debited")
text.normal("from your account.")
await require_confirm(ctx, text, ButtonRequestType.ConfirmOutput)
await confirm_asset_issuer(ctx, op.send_asset)
async def confirm_payment_op(ctx, op: StellarPaymentOp):
text = Text("Confirm operation", ui.ICON_CONFIRM, icon_color=ui.GREEN)
text.bold("Pay %s" % format_amount(op.amount, ticker=False))
text.bold("%s to:" % format_asset_code(op.asset))
text.mono(*split(trim_to_rows(op.destination_account, 3)))
await require_confirm(ctx, text, ButtonRequestType.ConfirmOutput)
await confirm_asset_issuer(ctx, op.asset)
async def confirm_set_options_op(ctx, op: StellarSetOptionsOp):
if op.inflation_destination_account:
text = Text("Confirm operation", ui.ICON_CONFIRM, icon_color=ui.GREEN)
text.bold("Set Inflation Destination")
text.mono(*split(op.inflation_destination_account))
await require_confirm(ctx, text, ButtonRequestType.ConfirmOutput)
if op.clear_flags:
t = _format_flags(op.clear_flags)
text = Text("Confirm operation", ui.ICON_CONFIRM, icon_color=ui.GREEN)
text.bold("Clear Flags")
text.mono(*t)
await require_confirm(ctx, text, ButtonRequestType.ConfirmOutput)
if op.set_flags:
t = _format_flags(op.set_flags)
text = Text("Confirm operation", ui.ICON_CONFIRM, icon_color=ui.GREEN)
text.bold("Set Flags")
text.mono(*t)
await require_confirm(ctx, text, ButtonRequestType.ConfirmOutput)
thresholds = _format_thresholds(op)
if thresholds:
text = Text("Confirm operation", ui.ICON_CONFIRM, icon_color=ui.GREEN)
text.bold("Account Thresholds")
text.mono(*thresholds)
await require_confirm(ctx, text, ButtonRequestType.ConfirmOutput)
if op.home_domain:
text = Text("Confirm operation", ui.ICON_CONFIRM, icon_color=ui.GREEN)
text.bold("Home Domain")
text.mono(*split(op.home_domain))
await require_confirm(ctx, text, ButtonRequestType.ConfirmOutput)
if op.signer_type is not None:
if op.signer_weight > 0:
t = "Add Signer (%s)"
else:
t = "Remove Signer (%s)"
if op.signer_type == consts.SIGN_TYPE_ACCOUNT:
text = Text("Confirm operation", ui.ICON_CONFIRM, icon_color=ui.GREEN)
text.bold(t % "acc")
text.mono(*split(op.signer_key))
await require_confirm(ctx, text, ButtonRequestType.ConfirmOutput)
elif op.signer_type in (consts.SIGN_TYPE_PRE_AUTH, consts.SIGN_TYPE_HASH):
if op.signer_type == consts.SIGN_TYPE_PRE_AUTH:
signer_type = "auth"
else:
signer_type = "hash"
text = Text("Confirm operation", ui.ICON_CONFIRM, icon_color=ui.GREEN)
text.bold(t % signer_type)
text.mono(*split(hexlify(op.signer_key).decode()))
await require_confirm(ctx, text, ButtonRequestType.ConfirmOutput)
else:
raise ProcessError("Stellar: invalid signer type")
def _format_thresholds(op: StellarSetOptionsOp) -> tuple:
text = ()
if op.master_weight is not None:
text += ("Master Weight: %d" % op.master_weight,)
if op.low_threshold is not None:
text += ("Low: %d" % op.low_threshold,)
if op.medium_threshold is not None:
text += ("Medium: %d" % op.medium_threshold,)
if op.high_threshold is not None:
text += ("High: %d" % op.high_threshold,)
return text
def _format_flags(flags: int) -> tuple:
if flags > consts.FLAGS_MAX_SIZE:
raise ProcessError("Stellar: invalid flags")
text = ()
if flags & consts.FLAG_AUTH_REQUIRED:
text += ("AUTH_REQUIRED",)
if flags & consts.FLAG_AUTH_REVOCABLE:
text += ("AUTH_REVOCABLE",)
if flags & consts.FLAG_AUTH_IMMUTABLE:
text += ("AUTH_IMMUTABLE",)
return text
def format_asset_code(asset: StellarAssetType) -> str:
if asset is None or asset.type == consts.ASSET_TYPE_NATIVE:
return "XLM (native)"
return asset.code
async def confirm_asset_issuer(ctx, asset: StellarAssetType):
if asset is None or asset.type == consts.ASSET_TYPE_NATIVE:
return
text = Text("Confirm issuer", ui.ICON_CONFIRM, icon_color=ui.GREEN)
text.bold("%s issuer:" % asset.code)
text.mono(*split(asset.issuer))
await require_confirm(ctx, text, ButtonRequestType.ConfirmOutput)

@ -0,0 +1,166 @@
from trezor.messages.StellarAccountMergeOp import StellarAccountMergeOp
from trezor.messages.StellarAllowTrustOp import StellarAllowTrustOp
from trezor.messages.StellarAssetType import StellarAssetType
from trezor.messages.StellarBumpSequenceOp import StellarBumpSequenceOp
from trezor.messages.StellarChangeTrustOp import StellarChangeTrustOp
from trezor.messages.StellarCreateAccountOp import StellarCreateAccountOp
from trezor.messages.StellarCreatePassiveOfferOp import StellarCreatePassiveOfferOp
from trezor.messages.StellarManageDataOp import StellarManageDataOp
from trezor.messages.StellarManageOfferOp import StellarManageOfferOp
from trezor.messages.StellarPathPaymentOp import StellarPathPaymentOp
from trezor.messages.StellarPaymentOp import StellarPaymentOp
from trezor.messages.StellarSetOptionsOp import StellarSetOptionsOp
from trezor.wire import ProcessError
from apps.stellar import consts, writers
def serialize_account_merge_op(w, msg: StellarAccountMergeOp):
writers.write_pubkey(w, msg.destination_account)
def serialize_allow_trust_op(w, msg: StellarAllowTrustOp):
# trustor account (the account being allowed to access the asset)
writers.write_pubkey(w, msg.trusted_account)
writers.write_uint32(w, msg.asset_type)
_serialize_asset_code(w, msg.asset_type, msg.asset_code)
writers.write_bool(w, msg.is_authorized)
def serialize_bump_sequence_op(w, msg: StellarBumpSequenceOp):
writers.write_uint64(w, msg.bump_to)
def serialize_change_trust_op(w, msg: StellarChangeTrustOp):
_serialize_asset(w, msg.asset)
writers.write_uint64(w, msg.limit)
def serialize_create_account_op(w, msg: StellarCreateAccountOp):
writers.write_pubkey(w, msg.new_account)
writers.write_uint64(w, msg.starting_balance)
def serialize_create_passive_offer_op(w, msg: StellarCreatePassiveOfferOp):
_serialize_asset(w, msg.selling_asset)
_serialize_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)
def serialize_manage_data_op(w, msg: StellarManageDataOp):
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))
if msg.value:
writers.write_uint32(w, len(msg.value))
writers.write_bytes(w, msg.value)
def serialize_manage_offer_op(w, msg: StellarManageOfferOp):
_serialize_asset(w, msg.selling_asset)
_serialize_asset(w, msg.buying_asset)
writers.write_uint64(w, msg.amount) # amount to sell
writers.write_uint32(w, msg.price_n) # numerator
writers.write_uint32(w, msg.price_d) # denominator
writers.write_uint64(w, msg.offer_id)
def serialize_path_payment_op(w, msg: StellarPathPaymentOp):
_serialize_asset(w, msg.send_asset)
writers.write_uint64(w, msg.send_max)
writers.write_pubkey(w, msg.destination_account)
_serialize_asset(w, msg.destination_asset)
writers.write_uint64(w, msg.destination_amount)
writers.write_uint32(w, len(msg.paths))
for p in msg.paths:
_serialize_asset(w, p)
def serialize_payment_op(w, msg: StellarPaymentOp):
writers.write_pubkey(w, msg.destination_account)
_serialize_asset(w, msg.asset)
writers.write_uint64(w, msg.amount)
def serialize_set_options_op(w, msg: StellarSetOptionsOp):
# inflation destination
writers.write_bool(w, bool(msg.inflation_destination_account))
if msg.inflation_destination_account:
writers.write_pubkey(w, msg.inflation_destination_account)
# clear flags
writers.write_bool(w, bool(msg.clear_flags))
if msg.clear_flags:
writers.write_uint32(w, msg.clear_flags)
# set flags
writers.write_bool(w, bool(msg.set_flags))
if msg.set_flags:
writers.write_uint32(w, msg.set_flags)
# account thresholds
writers.write_bool(w, bool(msg.master_weight))
if msg.master_weight:
writers.write_uint32(w, msg.master_weight)
writers.write_bool(w, bool(msg.low_threshold))
if msg.low_threshold:
writers.write_uint32(w, msg.low_threshold)
writers.write_bool(w, bool(msg.medium_threshold))
if msg.medium_threshold:
writers.write_uint32(w, msg.medium_threshold)
writers.write_bool(w, bool(msg.high_threshold))
if msg.high_threshold:
writers.write_uint32(w, msg.high_threshold)
# home domain
writers.write_bool(w, bool(msg.home_domain))
if msg.home_domain:
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)
# signer
writers.write_bool(w, bool(msg.signer_type))
if msg.signer_type:
# signer type
writers.write_uint32(w, msg.signer_type)
writers.write_bytes(w, msg.signer_key)
writers.write_uint32(w, msg.signer_weight)
def serialize_account(w, source_account: str):
if source_account is None:
writers.write_bool(w, False)
return
writers.write_pubkey(w, source_account)
def _serialize_asset_code(w, asset_type: int, asset_code: str):
code = bytearray(asset_code)
if asset_type == consts.ASSET_TYPE_NATIVE:
return # nothing is needed
elif asset_type == consts.ASSET_TYPE_ALPHANUM4:
# pad with zeros to 4 chars
writers.write_bytes(w, code + bytearray([0] * (4 - len(code))))
elif asset_type == consts.ASSET_TYPE_ALPHANUM12:
# pad with zeros to 12 chars
writers.write_bytes(w, code + bytearray([0] * (12 - len(code))))
else:
raise ProcessError("Stellar: invalid asset type")
def _serialize_asset(w, asset: StellarAssetType):
if asset is None:
writers.write_uint32(w, 0)
return
writers.write_uint32(w, asset.type)
_serialize_asset_code(w, asset.type, asset.code)
writers.write_pubkey(w, asset.issuer)

@ -0,0 +1,99 @@
from ubinascii import hexlify
from trezor.crypto.curve import ed25519
from trezor.crypto.hashlib import sha256
from trezor.messages.StellarSignedTx import StellarSignedTx
from trezor.messages.StellarSignTx import StellarSignTx
from trezor.messages.StellarTxOpRequest import StellarTxOpRequest
from trezor.wire import ProcessError
from apps.common import seed
from apps.stellar import consts, helpers, layout, writers
from apps.stellar.operations import operation
async def sign_tx(ctx, msg: StellarSignTx):
if msg.num_operations == 0:
raise ProcessError("Stellar: At least one operation is required")
node = await seed.derive_node(ctx, msg.address_n, consts.STELLAR_CURVE)
pubkey = seed.remove_ed25519_prefix(node.public_key())
w = bytearray()
await _init(ctx, w, pubkey, msg)
_timebounds(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(pubkey, signature)
async def _final(ctx, w: bytearray, msg: StellarSignTx):
# 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, w: bytearray, pubkey: bytes, msg: StellarSignTx):
network_passphrase_hash = sha256(msg.network_passphrase).digest()
writers.write_bytes(w, network_passphrase_hash)
writers.write_bytes(w, consts.TX_TYPE)
address = helpers.address_from_public_key(pubkey)
writers.write_pubkey(w, address)
if helpers.public_key_from_address(msg.source_account) != pubkey:
raise ProcessError("Stellar: source account does not match address_n")
writers.write_uint32(w, msg.fee)
writers.write_uint64(w, msg.sequence_number)
# confirm init
await layout.require_confirm_init(ctx, address, msg.network_passphrase)
def _timebounds(w: bytearray, start: int, end: int):
# timebounds are only present if timebounds_start or timebounds_end is non-zero
if start or end:
writers.write_bool(w, True)
# timebounds are sent as uint32s since that's all we can display, but they must be hashed as 64bit
writers.write_uint64(w, start)
writers.write_uint64(w, end)
else:
writers.write_bool(w, False)
async def _operations(ctx, w: bytearray, num_operations: int):
writers.write_uint32(w, num_operations)
for i in range(num_operations):
op = await ctx.call(StellarTxOpRequest(), *consts.op_wire_types)
await operation(ctx, w, op)
async def _memo(ctx, w: bytearray, msg: StellarSignTx):
writers.write_uint32(w, msg.memo_type)
if msg.memo_type == consts.MEMO_TYPE_NONE:
# nothing is serialized
memo_confirm_text = ""
elif msg.memo_type == consts.MEMO_TYPE_TEXT:
# Text: 4 bytes (size) + up to 28 bytes
if len(msg.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 == consts.MEMO_TYPE_ID:
# ID: 64 bit unsigned integer
writers.write_uint64(w, msg.memo_id)
memo_confirm_text = str(msg.memo_id)
elif msg.memo_type in (consts.MEMO_TYPE_HASH, consts.MEMO_TYPE_RETURN):
# Hash/Return: 32 byte hash
writers.write_bytes(w, bytearray(msg.memo_hash))
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)

@ -0,0 +1,38 @@
import ustruct
from .helpers import public_key_from_address
def write_uint32(w, n: int):
write_bytes(w, ustruct.pack(">L", n))
def write_uint64(w, n: int):
write_bytes(w, ustruct.pack(">Q", n))
def write_string(w, s: str):
write_uint32(w, len(s))
write_bytes(w, bytearray(s))
# if len isn't a multiple of 4, add padding bytes
reminder = len(s) % 4
if reminder:
write_bytes(w, bytearray([0] * (4 - reminder)))
def write_bytes(w, buf: bytearray):
w.extend(buf)
def write_bool(w, val: True):
if val:
write_uint32(w, 1)
else:
write_uint32(w, 0)
def write_pubkey(w, address: str):
# first 4 bytes of an address are the type, there's only one type (0)
write_uint32(w, 0)
pubkey = public_key_from_address(address)
write_bytes(w, bytearray(pubkey))

@ -15,6 +15,7 @@ import apps.wallet
import apps.ethereum
import apps.lisk
import apps.nem
import apps.stellar
if __debug__:
import apps.debug
@ -28,6 +29,7 @@ apps.wallet.boot()
apps.ethereum.boot()
apps.lisk.boot()
apps.nem.boot()
apps.stellar.boot()
if __debug__:
apps.debug.boot()
else:

@ -1,5 +1,6 @@
# Automatically generated by pb2py
# fmt: off
import protobuf as p

@ -1,5 +1,6 @@
# Automatically generated by pb2py
# fmt: off
import protobuf as p
if __debug__:
try:

@ -0,0 +1,38 @@
from common import *
from apps.stellar.helpers import address_from_public_key, public_key_from_address
from trezor.wire import ProcessError
class TestStellarAddressToPubkey(unittest.TestCase):
def test_address_to_pubkey(self):
self.assertEqual(public_key_from_address('GBOVKZBEM2YYLOCDCUXJ4IMRKHN4LCJAE7WEAEA2KF562XFAGDBOB64V'),
unhexlify('5d55642466b185b843152e9e219151dbc5892027ec40101a517bed5ca030c2e0'))
self.assertEqual(public_key_from_address('GCN2K2HG53AWX2SP5UHRPMJUUHLJF2XBTGSXROTPWRGAYJCDDP63J2U6'),
unhexlify('9ba568e6eec16bea4fed0f17b134a1d692eae199a578ba6fb44c0c24431bfdb4'))
def test_pubkey_to_address(self):
addr = address_from_public_key(unhexlify('5d55642466b185b843152e9e219151dbc5892027ec40101a517bed5ca030c2e0'))
self.assertEqual(addr, 'GBOVKZBEM2YYLOCDCUXJ4IMRKHN4LCJAE7WEAEA2KF562XFAGDBOB64V')
addr = address_from_public_key(unhexlify('9ba568e6eec16bea4fed0f17b134a1d692eae199a578ba6fb44c0c24431bfdb4'))
self.assertEqual(addr, 'GCN2K2HG53AWX2SP5UHRPMJUUHLJF2XBTGSXROTPWRGAYJCDDP63J2U6')
def test_both(self):
pubkey = unhexlify('dfcc77d08588601702e02de2dc603f5c5281bea23baa894ae3b3b4778e5bbe40')
self.assertEqual(public_key_from_address(address_from_public_key(pubkey)), pubkey)
pubkey = unhexlify('53214e6155469c32fb882b1b1d94930d5445a78202867b7ddc6a33ad42ff4464')
self.assertEqual(public_key_from_address(address_from_public_key(pubkey)), pubkey)
pubkey = unhexlify('5ed4690134e5ef79b290ea1e7a4b8f3b6b3bcf287463c18bfe36baa030e7efbd')
self.assertEqual(public_key_from_address(address_from_public_key(pubkey)), pubkey)
def test_invalid_address(self):
with self.assertRaises(ProcessError):
public_key_from_address('GCN2K2HG53AWX2SP5UHRPMJUUHLJF2XBTGSXROTPWRGAYJCDDP63J2AA') # invalid checksum
if __name__ == '__main__':
unittest.main()
Loading…
Cancel
Save