matejcik
3 years ago
committed by
matejcik
parent
dd655422f1
commit
aaa3ce6117
@ -1,64 +1,59 @@
|
||||
from micropython import const
|
||||
|
||||
from trezor.messages import MessageType
|
||||
from trezor import wire
|
||||
from trezor.messages.AuthorizeCoinJoin import AuthorizeCoinJoin
|
||||
|
||||
from apps.common import authorization
|
||||
|
||||
from .common import BIP32_WALLET_DEPTH
|
||||
|
||||
if False:
|
||||
from typing import Iterable
|
||||
from trezor.messages.AuthorizeCoinJoin import AuthorizeCoinJoin
|
||||
import protobuf
|
||||
from trezor.messages.GetOwnershipProof import GetOwnershipProof
|
||||
from trezor.messages.SignTx import SignTx
|
||||
from trezor.messages.TxInput import TxInput
|
||||
|
||||
from apps.common.coininfo import CoinInfo
|
||||
from apps.common.keychain import Keychain
|
||||
|
||||
_ROUND_ID_LEN = const(32)
|
||||
FEE_PER_ANONYMITY_DECIMALS = const(9)
|
||||
|
||||
|
||||
class CoinJoinAuthorization:
|
||||
def __init__(
|
||||
self, msg: AuthorizeCoinJoin, keychain: Keychain, coin: CoinInfo
|
||||
) -> None:
|
||||
self.coordinator = msg.coordinator
|
||||
self.remaining_fee = msg.max_total_fee
|
||||
self.fee_per_anonymity = msg.fee_per_anonymity or 0
|
||||
self.address_n = msg.address_n
|
||||
self.keychain = keychain
|
||||
self.coin = coin
|
||||
self.script_type = msg.script_type
|
||||
|
||||
def __del__(self) -> None:
|
||||
self.keychain.__del__()
|
||||
|
||||
def expected_wire_types(self) -> Iterable[int]:
|
||||
return (MessageType.SignTx, MessageType.GetOwnershipProof)
|
||||
def __init__(self, params: AuthorizeCoinJoin) -> None:
|
||||
self.params = params
|
||||
|
||||
def check_get_ownership_proof(self, msg: GetOwnershipProof) -> bool:
|
||||
# Check whether the current authorization matches the parameters of the request.
|
||||
# Check whether the current params matches the parameters of the request.
|
||||
return (
|
||||
len(msg.address_n) >= BIP32_WALLET_DEPTH
|
||||
and msg.address_n[:-BIP32_WALLET_DEPTH] == self.address_n
|
||||
and msg.coin_name == self.coin.coin_name
|
||||
and msg.script_type == self.script_type
|
||||
and msg.address_n[:-BIP32_WALLET_DEPTH] == self.params.address_n
|
||||
and msg.coin_name == self.params.coin_name
|
||||
and msg.script_type == self.params.script_type
|
||||
and len(msg.commitment_data) >= _ROUND_ID_LEN
|
||||
and msg.commitment_data[:-_ROUND_ID_LEN] == self.coordinator.encode()
|
||||
and msg.commitment_data[:-_ROUND_ID_LEN] == self.params.coordinator.encode()
|
||||
)
|
||||
|
||||
def check_sign_tx_input(self, txi: TxInput, coin: CoinInfo) -> bool:
|
||||
# Check whether the current input matches the parameters of the request.
|
||||
return (
|
||||
len(txi.address_n) >= BIP32_WALLET_DEPTH
|
||||
and txi.address_n[:-BIP32_WALLET_DEPTH] == self.address_n
|
||||
and coin.coin_name == self.coin.coin_name
|
||||
and txi.script_type == self.script_type
|
||||
and txi.address_n[:-BIP32_WALLET_DEPTH] == self.params.address_n
|
||||
and coin.coin_name == self.params.coin_name
|
||||
and txi.script_type == self.params.script_type
|
||||
)
|
||||
|
||||
def approve_sign_tx(self, msg: SignTx, fee: int) -> bool:
|
||||
if self.remaining_fee < fee or msg.coin_name != self.coin.coin_name:
|
||||
if self.params.max_total_fee < fee or msg.coin_name != self.params.coin_name:
|
||||
return False
|
||||
|
||||
self.remaining_fee -= fee
|
||||
self.params.max_total_fee -= fee
|
||||
authorization.set(self.params)
|
||||
return True
|
||||
|
||||
|
||||
def from_cached_message(auth_msg: protobuf.MessageType) -> CoinJoinAuthorization:
|
||||
if not isinstance(auth_msg, AuthorizeCoinJoin):
|
||||
raise wire.ProcessError("Appropriate params was not found")
|
||||
|
||||
return CoinJoinAuthorization(auth_msg)
|
||||
|
||||
@ -1,75 +1,71 @@
|
||||
from micropython import const
|
||||
|
||||
from trezor import ui
|
||||
from trezor import ui, wire
|
||||
from trezor.messages.AuthorizeCoinJoin import AuthorizeCoinJoin
|
||||
from trezor.messages.Success import Success
|
||||
from trezor.strings import format_amount
|
||||
from trezor.ui.layouts import confirm_action, confirm_coinjoin
|
||||
|
||||
from apps.base import set_authorization
|
||||
from apps.common import authorization
|
||||
from apps.common.paths import validate_path
|
||||
|
||||
from .authorization import FEE_PER_ANONYMITY_DECIMALS, CoinJoinAuthorization
|
||||
from .authorization import FEE_PER_ANONYMITY_DECIMALS
|
||||
from .common import BIP32_WALLET_DEPTH
|
||||
from .keychain import get_keychain_for_coin, validate_path_against_script_type
|
||||
from .keychain import validate_path_against_script_type, with_keychain
|
||||
from .sign_tx.layout import format_coin_amount
|
||||
|
||||
if False:
|
||||
from trezor import wire
|
||||
from apps.common.coininfo import CoinInfo
|
||||
from apps.common.keychain import Keychain
|
||||
|
||||
_MAX_COORDINATOR_LEN = const(18)
|
||||
|
||||
|
||||
async def authorize_coinjoin(ctx: wire.Context, msg: AuthorizeCoinJoin) -> Success:
|
||||
# We cannot use the @with_keychain decorator here, because we need the keychain
|
||||
# to survive the function exit. The ownership of the keychain is transferred to
|
||||
# the CoinJoinAuthorization object, which takes care of its destruction.
|
||||
keychain, coin = await get_keychain_for_coin(ctx, msg.coin_name)
|
||||
@with_keychain
|
||||
async def authorize_coinjoin(
|
||||
ctx: wire.Context, msg: AuthorizeCoinJoin, keychain: Keychain, coin: CoinInfo
|
||||
) -> Success:
|
||||
if len(msg.coordinator) > _MAX_COORDINATOR_LEN or not all(
|
||||
32 <= ord(x) <= 126 for x in msg.coordinator
|
||||
):
|
||||
raise wire.DataError("Invalid coordinator name.")
|
||||
|
||||
try:
|
||||
if len(msg.coordinator) > _MAX_COORDINATOR_LEN or not all(
|
||||
32 <= ord(x) <= 126 for x in msg.coordinator
|
||||
):
|
||||
raise wire.DataError("Invalid coordinator name.")
|
||||
if not msg.address_n:
|
||||
raise wire.DataError("Empty path not allowed.")
|
||||
|
||||
if not msg.address_n:
|
||||
raise wire.DataError("Empty path not allowed.")
|
||||
validation_path = msg.address_n + [0] * BIP32_WALLET_DEPTH
|
||||
await validate_path(
|
||||
ctx,
|
||||
keychain,
|
||||
validation_path,
|
||||
validate_path_against_script_type(
|
||||
coin, address_n=validation_path, script_type=msg.script_type
|
||||
),
|
||||
)
|
||||
|
||||
validation_path = msg.address_n + [0] * BIP32_WALLET_DEPTH
|
||||
await validate_path(
|
||||
ctx,
|
||||
keychain,
|
||||
validation_path,
|
||||
validate_path_against_script_type(
|
||||
coin, address_n=validation_path, script_type=msg.script_type
|
||||
),
|
||||
)
|
||||
await confirm_action(
|
||||
ctx,
|
||||
"coinjoin_coordinator",
|
||||
title="Authorize CoinJoin",
|
||||
description="Do you really want to take part in a CoinJoin transaction at:\n{}",
|
||||
description_param=msg.coordinator,
|
||||
description_param_font=ui.MONO,
|
||||
icon=ui.ICON_RECOVERY,
|
||||
)
|
||||
|
||||
await confirm_action(
|
||||
ctx,
|
||||
"coinjoin_coordinator",
|
||||
title="Authorize CoinJoin",
|
||||
description="Do you really want to take part in a CoinJoin transaction at:\n{}",
|
||||
description_param=msg.coordinator,
|
||||
description_param_font=ui.MONO,
|
||||
icon=ui.ICON_RECOVERY,
|
||||
if msg.fee_per_anonymity:
|
||||
fee_per_anonymity: str | None = format_amount(
|
||||
msg.fee_per_anonymity, FEE_PER_ANONYMITY_DECIMALS
|
||||
)
|
||||
|
||||
else:
|
||||
fee_per_anonymity = None
|
||||
if msg.fee_per_anonymity is not None:
|
||||
fee_per_anonymity = format_amount(
|
||||
msg.fee_per_anonymity, FEE_PER_ANONYMITY_DECIMALS
|
||||
)
|
||||
await confirm_coinjoin(
|
||||
ctx,
|
||||
fee_per_anonymity,
|
||||
format_coin_amount(msg.max_total_fee, coin, msg.amount_unit),
|
||||
)
|
||||
|
||||
set_authorization(CoinJoinAuthorization(msg, keychain, coin))
|
||||
await confirm_coinjoin(
|
||||
ctx,
|
||||
fee_per_anonymity,
|
||||
format_coin_amount(msg.max_total_fee, coin, msg.amount_unit),
|
||||
)
|
||||
|
||||
except BaseException:
|
||||
keychain.__del__()
|
||||
raise
|
||||
authorization.set(msg)
|
||||
|
||||
return Success(message="CoinJoin authorized")
|
||||
|
||||
@ -0,0 +1,53 @@
|
||||
import protobuf
|
||||
import storage.cache
|
||||
from trezor import messages, utils
|
||||
from trezor.messages import MessageType
|
||||
|
||||
if False:
|
||||
from typing import Iterable
|
||||
|
||||
WIRE_TYPES: dict[int, tuple[int, ...]] = {
|
||||
MessageType.AuthorizeCoinJoin: (MessageType.SignTx, MessageType.GetOwnershipProof),
|
||||
}
|
||||
|
||||
|
||||
def is_set() -> bool:
|
||||
return bool(storage.cache.get(storage.cache.APP_COMMON_AUTHORIZATION_TYPE))
|
||||
|
||||
|
||||
def set(auth_message: protobuf.MessageType) -> None:
|
||||
buffer = bytearray(protobuf.count_message(auth_message))
|
||||
writer = utils.BufferWriter(buffer)
|
||||
protobuf.dump_message(writer, auth_message)
|
||||
storage.cache.set(
|
||||
storage.cache.APP_COMMON_AUTHORIZATION_TYPE,
|
||||
auth_message.MESSAGE_WIRE_TYPE.to_bytes(2, "big"),
|
||||
)
|
||||
storage.cache.set(storage.cache.APP_COMMON_AUTHORIZATION_DATA, buffer)
|
||||
|
||||
|
||||
def get() -> protobuf.MessageType | None:
|
||||
stored_auth_type = storage.cache.get(storage.cache.APP_COMMON_AUTHORIZATION_TYPE)
|
||||
if not stored_auth_type:
|
||||
return None
|
||||
|
||||
msg_wire_type = int.from_bytes(stored_auth_type, "big")
|
||||
msg_type = messages.get_type(msg_wire_type)
|
||||
buffer = storage.cache.get(storage.cache.APP_COMMON_AUTHORIZATION_DATA)
|
||||
reader = utils.BufferReader(buffer)
|
||||
|
||||
return protobuf.load_message(reader, msg_type)
|
||||
|
||||
|
||||
def get_wire_types() -> Iterable[int]:
|
||||
stored_auth_type = storage.cache.get(storage.cache.APP_COMMON_AUTHORIZATION_TYPE)
|
||||
if not stored_auth_type:
|
||||
return ()
|
||||
|
||||
msg_wire_type = int.from_bytes(stored_auth_type, "big")
|
||||
return WIRE_TYPES.get(msg_wire_type, ())
|
||||
|
||||
|
||||
def clear() -> None:
|
||||
storage.cache.set(storage.cache.APP_COMMON_AUTHORIZATION_TYPE, b"")
|
||||
storage.cache.set(storage.cache.APP_COMMON_AUTHORIZATION_DATA, b"")
|
||||
Loading…
Reference in new issue