feat(core): Implement SLIP-0025 CoinJoin accounts.

andrewkozlik/account-naming
Andrew Kozlik 2 years ago
parent 9261f2ffce
commit 19d5c43ce5

@ -441,6 +441,7 @@ message DoPreauthorized {
* @start * @start
* @next SignTx * @next SignTx
* @next GetOwnershipProof * @next GetOwnershipProof
* @next GetPublicKey
*/ */
message PreauthorizedRequest { message PreauthorizedRequest {
} }

@ -0,0 +1 @@
Add SLIP-0025 CoinJoin accounts.

@ -2,7 +2,7 @@ from micropython import const
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
from trezor import utils, wire from trezor import utils, wire
from trezor.messages import AuthorizeCoinJoin from trezor.messages import AuthorizeCoinJoin, GetPublicKey
from apps.common import authorization from apps.common import authorization
@ -26,6 +26,10 @@ class CoinJoinAuthorization:
def __init__(self, params: AuthorizeCoinJoin) -> None: def __init__(self, params: AuthorizeCoinJoin) -> None:
self.params = params self.params = params
def check_get_public_key(self, msg: GetPublicKey) -> bool:
# Check whether the current params matches the parameters of the XPUB request.
return self.params.address_n == msg.address_n
def check_get_ownership_proof(self, msg: GetOwnershipProof) -> bool: def check_get_ownership_proof(self, msg: GetOwnershipProof) -> bool:
# Check whether the current params matches the parameters of the request. # Check whether the current params matches the parameters of the request.
coordinator = utils.empty_bytearray(1 + len(self.params.coordinator.encode())) coordinator = utils.empty_bytearray(1 + len(self.params.coordinator.encode()))

@ -2,17 +2,17 @@ from micropython import const
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
from trezor import ui, wire from trezor import ui, wire
from trezor.enums import ButtonRequestType from trezor.enums import ButtonRequestType, InputScriptType
from trezor.messages import AuthorizeCoinJoin, Success from trezor.messages import AuthorizeCoinJoin, Success
from trezor.strings import format_amount from trezor.strings import format_amount
from trezor.ui.layouts import confirm_action, confirm_coinjoin, confirm_metadata from trezor.ui.layouts import confirm_action, confirm_coinjoin, confirm_metadata
from apps.common import authorization, safety_checks from apps.common import authorization, safety_checks
from apps.common.paths import validate_path from apps.common.paths import PathSchema, validate_path
from .authorization import FEE_RATE_DECIMALS from .authorization import FEE_RATE_DECIMALS
from .common import BIP32_WALLET_DEPTH from .common import BIP32_WALLET_DEPTH
from .keychain import validate_path_against_script_type, with_keychain from .keychain import PATTERN_SLIP25, with_keychain
if TYPE_CHECKING: if TYPE_CHECKING:
from apps.common.coininfo import CoinInfo from apps.common.coininfo import CoinInfo
@ -47,6 +47,13 @@ async def authorize_coinjoin(
if not msg.address_n: if not msg.address_n:
raise wire.DataError("Empty path not allowed.") raise wire.DataError("Empty path not allowed.")
validation_path = msg.address_n + [0] * BIP32_WALLET_DEPTH
if (
not PathSchema.parse(PATTERN_SLIP25, coin.slip44).match(validation_path)
or msg.script_type != InputScriptType.SPENDTAPROOT
) and safety_checks.is_strict():
raise wire.DataError("Forbidden path.")
await confirm_action( await confirm_action(
ctx, ctx,
"coinjoin_coordinator", "coinjoin_coordinator",
@ -60,14 +67,12 @@ async def authorize_coinjoin(
max_fee_per_vbyte = format_amount(msg.max_fee_per_kvbyte, 3) max_fee_per_vbyte = format_amount(msg.max_fee_per_kvbyte, 3)
await confirm_coinjoin(ctx, coin.coin_name, msg.max_rounds, max_fee_per_vbyte) await confirm_coinjoin(ctx, coin.coin_name, msg.max_rounds, max_fee_per_vbyte)
validation_path = msg.address_n + [0] * BIP32_WALLET_DEPTH
await validate_path( await validate_path(
ctx, ctx,
keychain, keychain,
validation_path, validation_path,
validate_path_against_script_type( PathSchema.parse(PATTERN_SLIP25, coin.slip44).match(validation_path),
coin, address_n=validation_path, script_type=msg.script_type msg.script_type == InputScriptType.SPENDTAPROOT,
),
) )
if msg.max_fee_per_kvbyte > coin.maxfee_kb: if msg.max_fee_per_kvbyte > coin.maxfee_kb:

@ -2,21 +2,42 @@ from typing import TYPE_CHECKING
from trezor import wire from trezor import wire
from trezor.enums import InputScriptType from trezor.enums import InputScriptType
from trezor.messages import HDNodeType, PublicKey from trezor.messages import AuthorizeCoinJoin, HDNodeType, PublicKey
from trezor.ui.layouts import confirm_action
from apps.common import coininfo, paths from apps.common import coininfo, paths
from apps.common.keychain import get_keychain from apps.common.keychain import get_keychain
from . import authorization
if TYPE_CHECKING: if TYPE_CHECKING:
from trezor.messages import GetPublicKey from trezor.messages import GetPublicKey
_SLIP25_PURPOSE = 10025 | paths.HARDENED
async def get_public_key(ctx: wire.Context, msg: GetPublicKey) -> PublicKey: async def get_public_key(
ctx: wire.Context,
msg: GetPublicKey,
auth_msg: AuthorizeCoinJoin | None = None,
) -> PublicKey:
coin_name = msg.coin_name or "Bitcoin" coin_name = msg.coin_name or "Bitcoin"
script_type = msg.script_type or InputScriptType.SPENDADDRESS script_type = msg.script_type or InputScriptType.SPENDADDRESS
coin = coininfo.by_name(coin_name) coin = coininfo.by_name(coin_name)
curve_name = msg.ecdsa_curve_name or coin.curve_name curve_name = msg.ecdsa_curve_name or coin.curve_name
# Require confirmation to access SLIP25 paths unless already authorized.
if auth_msg:
if not authorization.from_cached_message(auth_msg).check_get_public_key(msg):
raise wire.ProcessError("Unauthorized operation")
elif msg.address_n and msg.address_n[0] == _SLIP25_PURPOSE:
await confirm_action(
ctx,
"confirm_coinjoin_xpub",
title="CoinJoin XPUB",
description="Do you want to allow access to your CoinJoin account?",
)
keychain = await get_keychain(ctx, curve_name, [paths.AlwaysMatchingSchema]) keychain = await get_keychain(ctx, curve_name, [paths.AlwaysMatchingSchema])
node = keychain.derive(msg.address_n) node = keychain.derive(msg.address_n)

@ -4,6 +4,7 @@ from typing import TYPE_CHECKING
from trezor import wire from trezor import wire
from trezor.enums import InputScriptType from trezor.enums import InputScriptType
from trezor.messages import AuthorizeCoinJoin, SignTx
from apps.common import coininfo from apps.common import coininfo
from apps.common.keychain import get_keychain from apps.common.keychain import get_keychain
@ -19,13 +20,11 @@ if TYPE_CHECKING:
from trezor.protobuf import MessageType from trezor.protobuf import MessageType
from trezor.messages import ( from trezor.messages import (
AuthorizeCoinJoin,
GetAddress, GetAddress,
GetOwnershipId, GetOwnershipId,
GetOwnershipProof, GetOwnershipProof,
GetPublicKey, GetPublicKey,
SignMessage, SignMessage,
SignTx,
VerifyMessage, VerifyMessage,
) )
@ -66,6 +65,10 @@ PATTERN_BIP49 = "m/49'/coin_type'/account'/change/address_index"
PATTERN_BIP84 = "m/84'/coin_type'/account'/change/address_index" PATTERN_BIP84 = "m/84'/coin_type'/account'/change/address_index"
# BIP-86 for taproot: https://github.com/bitcoin/bips/blob/master/bip-0086.mediawiki # BIP-86 for taproot: https://github.com/bitcoin/bips/blob/master/bip-0086.mediawiki
PATTERN_BIP86 = "m/86'/coin_type'/account'/change/address_index" PATTERN_BIP86 = "m/86'/coin_type'/account'/change/address_index"
# SLIP-25 for CoinJoin: https://github.com/satoshilabs/slips/blob/master/slip-0025.md
# Only account=0 and script_type=1 are supported for now.
PATTERN_SLIP25 = "m/10025'/coin_type'/0'/1'/change/address_index"
PATTERN_SLIP25_EXTERNAL = "m/10025'/coin_type'/0'/1'/0/address_index"
# compatibility patterns, will be removed in the future # compatibility patterns, will be removed in the future
PATTERN_GREENADDRESS_A = "m/[1,4]/address_index" PATTERN_GREENADDRESS_A = "m/[1,4]/address_index"
@ -151,13 +154,16 @@ def validate_path_against_script_type(
elif coin.taproot and script_type == InputScriptType.SPENDTAPROOT: elif coin.taproot and script_type == InputScriptType.SPENDTAPROOT:
patterns.append(PATTERN_BIP86) patterns.append(PATTERN_BIP86)
patterns.append(PATTERN_SLIP25)
return any( return any(
PathSchema.parse(pattern, coin.slip44).match(address_n) for pattern in patterns PathSchema.parse(pattern, coin.slip44).match(address_n) for pattern in patterns
) )
def get_schemas_for_coin(coin: coininfo.CoinInfo) -> Iterable[PathSchema]: def get_schemas_for_coin(
coin: coininfo.CoinInfo, allow_slip25_internal: bool = False
) -> Iterable[PathSchema]:
# basic patterns # basic patterns
patterns = [ patterns = [
PATTERN_BIP44, PATTERN_BIP44,
@ -205,6 +211,10 @@ def get_schemas_for_coin(coin: coininfo.CoinInfo) -> Iterable[PathSchema]:
# taproot patterns # taproot patterns
if coin.taproot: if coin.taproot:
patterns.append(PATTERN_BIP86) patterns.append(PATTERN_BIP86)
if allow_slip25_internal:
patterns.append(PATTERN_SLIP25)
else:
patterns.append(PATTERN_SLIP25_EXTERNAL)
schemas = [PathSchema.parse(pattern, coin.slip44) for pattern in patterns] schemas = [PathSchema.parse(pattern, coin.slip44) for pattern in patterns]
@ -234,10 +244,12 @@ def get_coin_by_name(coin_name: str | None) -> coininfo.CoinInfo:
async def get_keychain_for_coin( async def get_keychain_for_coin(
ctx: wire.Context, coin_name: str | None ctx: wire.Context,
coin_name: str | None,
allow_slip25_internal: bool = False,
) -> tuple[Keychain, coininfo.CoinInfo]: ) -> tuple[Keychain, coininfo.CoinInfo]:
coin = get_coin_by_name(coin_name) coin = get_coin_by_name(coin_name)
schemas = get_schemas_for_coin(coin) schemas = get_schemas_for_coin(coin, allow_slip25_internal)
slip21_namespaces = [[b"SLIP-0019"], [b"SLIP-0024"]] slip21_namespaces = [[b"SLIP-0019"], [b"SLIP-0024"]]
keychain = await get_keychain(ctx, coin.curve_name, schemas, slip21_namespaces) keychain = await get_keychain(ctx, coin.curve_name, schemas, slip21_namespaces)
return keychain, coin return keychain, coin
@ -249,7 +261,13 @@ def with_keychain(func: HandlerWithCoinInfo[MsgOut]) -> Handler[MsgIn, MsgOut]:
msg: MsgIn, msg: MsgIn,
auth_msg: MessageType | None = None, auth_msg: MessageType | None = None,
) -> MsgOut: ) -> MsgOut:
keychain, coin = await get_keychain_for_coin(ctx, msg.coin_name) # Allow access to the SLIP25 internal chain (CoinJoin) for SignTx and preauthorized CoinJoin operations.
allow_slip25_internal = SignTx.is_type_of(msg) or (
auth_msg is not None and AuthorizeCoinJoin.is_type_of(auth_msg)
)
keychain, coin = await get_keychain_for_coin(
ctx, msg.coin_name, allow_slip25_internal
)
if auth_msg: if auth_msg:
auth_obj = authorization.from_cached_message(auth_msg) auth_obj = authorization.from_cached_message(auth_msg)
return await func(ctx, msg, keychain, coin, auth_obj) return await func(ctx, msg, keychain, coin, auth_obj)

@ -6,7 +6,11 @@ from trezor.enums import MessageType
from trezor.utils import ensure from trezor.utils import ensure
WIRE_TYPES: dict[int, tuple[int, ...]] = { WIRE_TYPES: dict[int, tuple[int, ...]] = {
MessageType.AuthorizeCoinJoin: (MessageType.SignTx, MessageType.GetOwnershipProof), MessageType.AuthorizeCoinJoin: (
MessageType.SignTx,
MessageType.GetOwnershipProof,
MessageType.GetPublicKey,
),
} }

Loading…
Cancel
Save