core/sign_tx: Move Bitcoin class to bitcoin.py.

pull/985/head
Andrew Kozlik 4 years ago committed by Andrew Kozlik
parent a07e125793
commit be7e98aa47

@ -7,13 +7,14 @@ from trezor.messages.TxRequest import TxRequest
from apps.common import coins, paths, seed
from apps.wallet.sign_tx import (
addresses,
bitcoin,
common,
helpers,
layout,
multisig,
progress,
scripts,
segwit_bip143,
signing,
)
if not utils.BITCOIN_ONLY:
@ -27,13 +28,13 @@ async def sign_tx(ctx: wire.Context, msg: SignTx, keychain: seed.Keychain) -> Tx
coin_name = msg.coin_name if msg.coin_name is not None else "Bitcoin"
coin = coins.by_name(coin_name)
if not utils.BITCOIN_ONLY and coin.decred:
coinsig = decred.Decred() # type: signing.Bitcoin
coinsig = decred.Decred() # type: bitcoin.Bitcoin
elif not utils.BITCOIN_ONLY and coin.overwintered:
coinsig = zcash.Overwintered()
elif not utils.BITCOIN_ONLY and coin_name not in ("Bitcoin", "Regtest", "Testnet"):
coinsig = bitcoinlike.Bitcoinlike()
else:
coinsig = signing.Bitcoin()
coinsig = bitcoin.Bitcoin()
signer = coinsig.signer(msg, keychain, coin)
@ -42,7 +43,7 @@ async def sign_tx(ctx: wire.Context, msg: SignTx, keychain: seed.Keychain) -> Tx
try:
req = signer.send(res)
except (
signing.SigningError,
common.SigningError,
multisig.MultisigError,
addresses.AddressError,
scripts.ScriptsError,

@ -2,8 +2,7 @@ import gc
from micropython import const
from trezor import utils
from trezor.crypto import base58, bip32, der
from trezor.crypto.curve import secp256k1
from trezor.crypto import base58
from trezor.crypto.hashlib import sha256
from trezor.messages import FailureType, InputScriptType, OutputScriptType
from trezor.messages.SignTx import SignTx
@ -26,13 +25,12 @@ from apps.wallet.sign_tx import (
tx_weight,
writers,
)
from apps.wallet.sign_tx.common import SigningError, ecdsa_sign
from apps.wallet.sign_tx.matchcheck import MultisigFingerprintChecker, WalletPathChecker
if False:
from typing import Dict, Union
# the number of bip32 levels used in a wallet (chain and address)
_BIP32_WALLET_DEPTH = const(2)
# the chain id used for change
_BIP32_CHANGE_CHAIN = const(1)
@ -44,79 +42,6 @@ _BIP32_MAX_LAST_ELEMENT = const(1000000)
_MAX_SERIALIZED_CHUNK_SIZE = const(2048)
class SigningError(ValueError):
pass
class MatchChecker:
"""
MatchCheckers are used to identify the change-output in a transaction. An output is a change-output
if it has certain matching attributes with all inputs.
1. When inputs are first processed, add_input() is called on each one to determine if they all match.
2. Outputs are tested using output_matches() to tell whether they are admissible as a change-output.
3. Before signing each input, check_input() is used to ensure that the attribute has not changed.
"""
MISMATCH = object()
UNDEFINED = object()
def __init__(self) -> None:
self.attribute = self.UNDEFINED # type: object
self.read_only = False # Failsafe to ensure that add_input() is not accidentally called after output_matches().
def attribute_from_tx(self, txio: Union[TxInputType, TxOutputType]) -> object:
# Return the attribute from the txio, which is to be used for matching.
# If the txio is invalid for matching, then return an object which
# evaluates as a boolean False.
raise NotImplementedError
def add_input(self, txi: TxInputType) -> None:
ensure(not self.read_only)
if self.attribute is self.MISMATCH:
return # There was a mismatch in previous inputs.
added_attribute = self.attribute_from_tx(txi)
if not added_attribute:
self.attribute = self.MISMATCH # The added input is invalid for matching.
elif self.attribute is self.UNDEFINED:
self.attribute = added_attribute # This is the first input.
elif self.attribute != added_attribute:
self.attribute = self.MISMATCH
def check_input(self, txi: TxInputType) -> None:
if self.attribute is self.MISMATCH:
return # There was already a mismatch when adding inputs, ignore it now.
# All added inputs had a matching attribute, allowing a change-output.
# Ensure that this input still has the same attribute.
if self.attribute != self.attribute_from_tx(txi):
raise SigningError(
FailureType.ProcessError, "Transaction has changed during signing"
)
def output_matches(self, txo: TxOutputType) -> bool:
self.read_only = True
if self.attribute is self.MISMATCH:
return False
return self.attribute_from_tx(txo) == self.attribute
class WalletPathChecker(MatchChecker):
def attribute_from_tx(self, txio: Union[TxInputType, TxOutputType]) -> object:
if not txio.address_n:
return None
return txio.address_n[:-_BIP32_WALLET_DEPTH]
class MultisigFingerprintChecker(MatchChecker):
def attribute_from_tx(self, txio: Union[TxInputType, TxOutputType]) -> object:
if not txio.multisig:
return None
return multisig.multisig_fingerprint(txio.multisig)
# Transaction signing
# ===
# see https://github.com/trezor/trezor-mcu/blob/master/firmware/signing.c#L84
@ -189,10 +114,11 @@ class Bitcoin:
# legacy inputs in Step 4.
self.h_confirmed = self.create_hash_writer() # not a real tx hash
self.init_hash143()
# BIP-0143 transaction hashing
self.hash143 = self.create_hash143()
def init_hash143(self) -> None:
self.hash143 = segwit_bip143.Bip143() # BIP-0143 transaction hashing
def create_hash143(self) -> segwit_bip143.Bip143:
return segwit_bip143.Bip143()
def create_hash_writer(self) -> utils.HashWriter:
return utils.HashWriter(sha256())
@ -275,8 +201,10 @@ class Bitcoin:
await helpers.confirm_foreign_address(txi.address_n)
if input_is_segwit(txi):
self.segwit[i] = True
await self.process_segwit_input(i, txi)
elif input_is_nonsegwit(txi):
self.segwit[i] = False
await self.process_nonsegwit_input(i, txi)
else:
raise SigningError(FailureType.DataError, "Wrong input script type")
@ -284,12 +212,10 @@ class Bitcoin:
async def process_segwit_input(self, i: int, txi: TxInputType) -> None:
if not txi.amount:
raise SigningError(FailureType.DataError, "Segwit input without amount")
self.segwit[i] = True
self.bip143_in += txi.amount
self.total_in += txi.amount
async def process_nonsegwit_input(self, i: int, txi: TxInputType) -> None:
self.segwit[i] = False
self.total_in += await self.get_prevtx_output_value(
txi.prev_hash, txi.prev_index
)
@ -384,6 +310,7 @@ class Bitcoin:
txi_sign = txi
self.wallet_path.check_input(txi_sign)
self.multisig_fingerprint.check_input(txi_sign)
# NOTE: wallet_path is checked in write_tx_input_check()
node = self.keychain.derive(txi.address_n, self.coin.curve_name)
key_sign_pub = node.public_key()
# for the signing process the script_sig is equal
@ -644,9 +571,3 @@ def input_is_segwit(txi: TxInputType) -> bool:
def input_is_nonsegwit(txi: TxInputType) -> bool:
return txi.script_type in helpers.NONSEGWIT_INPUT_SCRIPT_TYPES
def ecdsa_sign(node: bip32.HDNode, digest: bytes) -> bytes:
sig = secp256k1.sign(node.private_key(), digest)
sigder = der.encode_seq((sig[1:33], sig[33:65]))
return sigder

@ -8,18 +8,18 @@ from trezor.messages.TransactionType import TransactionType
from trezor.messages.TxInputType import TxInputType
from trezor.messages.TxOutputType import TxOutputType
from apps.wallet.sign_tx import addresses, helpers, multisig, signing, writers
from apps.wallet.sign_tx import addresses, helpers, multisig, writers
from apps.wallet.sign_tx.bitcoin import Bitcoin
from apps.wallet.sign_tx.common import SigningError, ecdsa_sign
if False:
from typing import Union
class Bitcoinlike(signing.Bitcoin):
class Bitcoinlike(Bitcoin):
async def process_segwit_input(self, i: int, txi: TxInputType) -> None:
if not self.coin.segwit:
raise signing.SigningError(
FailureType.DataError, "Segwit not enabled on this coin"
)
raise SigningError(FailureType.DataError, "Segwit not enabled on this coin")
await super().process_segwit_input(i, txi)
async def process_nonsegwit_input(self, i: int, txi: TxInputType) -> None:
@ -30,9 +30,7 @@ class Bitcoinlike(signing.Bitcoin):
async def process_bip143_input(self, i: int, txi: TxInputType) -> None:
if not txi.amount:
raise signing.SigningError(
FailureType.DataError, "Expected input with amount"
)
raise SigningError(FailureType.DataError, "Expected input with amount")
self.segwit[i] = False
self.bip143_in += txi.amount
self.total_in += txi.amount
@ -54,7 +52,7 @@ class Bitcoinlike(signing.Bitcoin):
or txi_sign.script_type == InputScriptType.SPENDMULTISIG
)
if not is_bip143 or txi_sign.amount > self.bip143_in:
raise signing.SigningError(
raise SigningError(
FailureType.ProcessError, "Transaction has changed during signing"
)
self.bip143_in -= txi_sign.amount
@ -73,7 +71,7 @@ class Bitcoinlike(signing.Bitcoin):
if txi_sign.multisig:
multisig.multisig_pubkey_index(txi_sign.multisig, key_sign_pub)
signature = signing.ecdsa_sign(key_sign, self.hash143_hash)
signature = ecdsa_sign(key_sign, self.hash143_hash)
# serialize input with correct signature
gc.collect()
@ -100,7 +98,7 @@ class Bitcoinlike(signing.Bitcoin):
elif version == cashaddr.ADDRESS_TYPE_P2SH:
version = self.coin.address_type_p2sh
else:
raise signing.SigningError("Unknown cashaddr address type")
raise SigningError("Unknown cashaddr address type")
return bytes([version]) + data
else:
return super().get_raw_address(txo)

@ -0,0 +1,12 @@
from trezor.crypto import bip32, der
from trezor.crypto.curve import secp256k1
class SigningError(ValueError):
pass
def ecdsa_sign(node: bip32.HDNode, digest: bytes) -> bytes:
sig = secp256k1.sign(node.private_key(), digest)
sigder = der.encode_seq((sig[1:33], sig[33:65]))
return sigder

@ -12,8 +12,9 @@ from trezor.utils import HashWriter, ensure
from apps.common import coininfo, seed
from apps.wallet.sign_tx import addresses, helpers, multisig, progress, scripts, writers
from apps.wallet.sign_tx.bitcoin import Bitcoin
from apps.wallet.sign_tx.common import SigningError, ecdsa_sign
from apps.wallet.sign_tx.segwit_bip143 import Bip143
from apps.wallet.sign_tx.signing import Bitcoin, SigningError, ecdsa_sign
DECRED_SERIALIZE_FULL = const(0 << 16)
DECRED_SERIALIZE_NO_WITNESS = const(1 << 16)
@ -64,8 +65,8 @@ class Decred(Bitcoin):
ensure(coin.decred)
super().initialize(tx, keychain, coin)
def init_hash143(self) -> None:
self.hash143 = DecredPrefixHasher(self.tx) # pseudo BIP-0143 prefix hashing
def create_hash143(self) -> Bip143:
return DecredPrefixHasher(self.tx) # pseudo BIP-0143 prefix hashing
def create_hash_writer(self) -> HashWriter:
return HashWriter(blake256())

@ -16,7 +16,7 @@ from trezor.messages.TxOutputBinType import TxOutputBinType
from trezor.messages.TxOutputType import TxOutputType
from trezor.messages.TxRequest import TxRequest
from .signing import SigningError
from .common import SigningError
from .writers import TX_HASH_SIZE
from apps.common.coininfo import CoinInfo

@ -0,0 +1,85 @@
from micropython import const
from trezor.messages import FailureType
from trezor.messages.TxInputType import TxInputType
from trezor.messages.TxOutputType import TxOutputType
from trezor.utils import ensure
from apps.wallet.sign_tx import multisig
from apps.wallet.sign_tx.common import SigningError
if False:
from typing import Union
# the number of bip32 levels used in a wallet (chain and address)
_BIP32_WALLET_DEPTH = const(2)
class MatchChecker:
"""
MatchCheckers are used to identify the change-output in a transaction. An output is a change-output
if it has certain matching attributes with all inputs.
1. When inputs are first processed, add_input() is called on each one to determine if they all match.
2. Outputs are tested using output_matches() to tell whether they are admissible as a change-output.
3. Before signing each input, check_input() is used to ensure that the attribute has not changed.
"""
MISMATCH = object()
UNDEFINED = object()
def __init__(self) -> None:
self.attribute = self.UNDEFINED # type: object
self.read_only = False # Failsafe to ensure that add_input() is not accidentally called after output_matches().
def attribute_from_tx(self, txio: Union[TxInputType, TxOutputType]) -> object:
# Return the attribute from the txio, which is to be used for matching.
# If the txio is invalid for matching, then return an object which
# evaluates as a boolean False.
raise NotImplementedError
def add_input(self, txi: TxInputType) -> None:
ensure(not self.read_only)
if self.attribute is self.MISMATCH:
return # There was a mismatch in previous inputs.
added_attribute = self.attribute_from_tx(txi)
if not added_attribute:
self.attribute = self.MISMATCH # The added input is invalid for matching.
elif self.attribute is self.UNDEFINED:
self.attribute = added_attribute # This is the first input.
elif self.attribute != added_attribute:
self.attribute = self.MISMATCH
def check_input(self, txi: TxInputType) -> None:
if self.attribute is self.MISMATCH:
return # There was already a mismatch when adding inputs, ignore it now.
# All added inputs had a matching attribute, allowing a change-output.
# Ensure that this input still has the same attribute.
if self.attribute != self.attribute_from_tx(txi):
raise SigningError(
FailureType.ProcessError, "Transaction has changed during signing"
)
def output_matches(self, txo: TxOutputType) -> bool:
self.read_only = True
if self.attribute is self.MISMATCH:
return False
return self.attribute_from_tx(txo) == self.attribute
class WalletPathChecker(MatchChecker):
def attribute_from_tx(self, txio: Union[TxInputType, TxOutputType]) -> object:
if not txio.address_n:
return None
return txio.address_n[:-_BIP32_WALLET_DEPTH]
class MultisigFingerprintChecker(MatchChecker):
def attribute_from_tx(self, txio: Union[TxInputType, TxOutputType]) -> object:
if not txio.multisig:
return None
return multisig.multisig_fingerprint(txio.multisig)

@ -11,10 +11,10 @@ from trezor.utils import HashWriter, ensure
from apps.common.coininfo import CoinInfo
from apps.common.seed import Keychain
from apps.wallet.sign_tx.bitcoinlike import Bitcoinlike
from apps.wallet.sign_tx.common import SigningError
from apps.wallet.sign_tx.multisig import multisig_get_pubkeys
from apps.wallet.sign_tx.scripts import output_script_multisig, output_script_p2pkh
from apps.wallet.sign_tx.segwit_bip143 import Bip143
from apps.wallet.sign_tx.signing import SigningError
from apps.wallet.sign_tx.writers import (
TX_HASH_SIZE,
get_tx_hash,
@ -33,10 +33,6 @@ if False:
OVERWINTERED = const(0x80000000)
class ZcashError(ValueError):
pass
def derive_script_code(txi: TxInputType, pubkeyhash: bytes) -> bytearray:
if txi.multisig:
@ -49,7 +45,7 @@ def derive_script_code(txi: TxInputType, pubkeyhash: bytes) -> bytearray:
return output_script_p2pkh(pubkeyhash)
else:
raise ZcashError(
raise SigningError(
FailureType.DataError, "Unknown input script type for zip143 script code"
)
@ -188,13 +184,13 @@ class Overwintered(Bitcoinlike):
ensure(coin.overwintered)
super().initialize(tx, keychain, coin)
def init_hash143(self) -> None:
def create_hash143(self) -> Bip143:
if self.tx.version == 3:
branch_id = self.tx.branch_id or 0x5BA81B19 # Overwinter
self.hash143 = Zip143(branch_id) # ZIP-0143 transaction hashing
return Zip143(branch_id) # ZIP-0143 transaction hashing
elif self.tx.version == 4:
branch_id = self.tx.branch_id or 0x76B809BB # Sapling
self.hash143 = Zip243(branch_id) # ZIP-0243 transaction hashing
return Zip243(branch_id) # ZIP-0243 transaction hashing
else:
raise SigningError(
FailureType.DataError,

Loading…
Cancel
Save