From 2b74513e49b0a351b544bcad4bd97b77bfd05a4a Mon Sep 17 00:00:00 2001 From: Andrew Kozlik Date: Wed, 8 Apr 2020 17:37:40 +0200 Subject: [PATCH] core: Fix mypy. --- core/src/apps/wallet/sign_tx/__init__.py | 12 ++++-- core/src/apps/wallet/sign_tx/addresses.py | 19 +++++---- core/src/apps/wallet/sign_tx/bitcoinlike.py | 2 +- core/src/apps/wallet/sign_tx/decred.py | 26 +++++++------ core/src/apps/wallet/sign_tx/helpers.py | 26 ++++++------- core/src/apps/wallet/sign_tx/layout.py | 39 +++++++++++++------ core/src/apps/wallet/sign_tx/multisig.py | 4 +- core/src/apps/wallet/sign_tx/scripts.py | 29 ++++++++------ core/src/apps/wallet/sign_tx/segwit_bip143.py | 8 ++-- core/src/apps/wallet/sign_tx/signing.py | 8 ++-- core/src/apps/wallet/sign_tx/tx_weight.py | 10 ++--- core/src/apps/wallet/sign_tx/writers.py | 17 ++++---- core/src/apps/wallet/sign_tx/zcash.py | 8 ++-- core/src/trezor/crypto/bech32.py | 6 +-- core/src/trezor/utils.py | 3 -- 15 files changed, 122 insertions(+), 95 deletions(-) diff --git a/core/src/apps/wallet/sign_tx/__init__.py b/core/src/apps/wallet/sign_tx/__init__.py index fce35196f..1842d8b90 100644 --- a/core/src/apps/wallet/sign_tx/__init__.py +++ b/core/src/apps/wallet/sign_tx/__init__.py @@ -1,9 +1,10 @@ from trezor import utils, wire from trezor.messages.RequestType import TXFINISHED +from trezor.messages.SignTx import SignTx from trezor.messages.TxAck import TxAck from trezor.messages.TxRequest import TxRequest -from apps.common import coins, paths +from apps.common import coins, paths, seed from apps.wallet.sign_tx import ( addresses, helpers, @@ -18,12 +19,15 @@ from apps.wallet.sign_tx import ( if not utils.BITCOIN_ONLY: from apps.wallet.sign_tx import decred, bitcoinlike +if False: + from typing import Union -async def sign_tx(ctx, msg, keychain): + +async def sign_tx(ctx: wire.Context, msg: SignTx, keychain: seed.Keychain) -> TxRequest: 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() + coinsig = decred.Decred() # type: signing.Bitcoin elif not utils.BITCOIN_ONLY and coin.overwintered: coinsig = bitcoinlike.Overwintered() elif not utils.BITCOIN_ONLY and coin_name not in ("Bitcoin", "Regtest", "Testnet"): @@ -33,7 +37,7 @@ async def sign_tx(ctx, msg, keychain): signer = coinsig.signer(msg, keychain, coin) - res = None + res = None # type: Union[TxAck, bool] while True: try: req = signer.send(res) diff --git a/core/src/apps/wallet/sign_tx/addresses.py b/core/src/apps/wallet/sign_tx/addresses.py index 3bd83db1a..182ba4c68 100644 --- a/core/src/apps/wallet/sign_tx/addresses.py +++ b/core/src/apps/wallet/sign_tx/addresses.py @@ -3,6 +3,7 @@ from micropython import const from trezor.crypto import base58, bech32, cashaddr from trezor.crypto.hashlib import sha256 from trezor.messages import FailureType, InputScriptType +from trezor.messages.MultisigRedeemScriptType import MultisigRedeemScriptType from trezor.utils import ensure from apps.common import HARDENED, address_type, paths @@ -15,6 +16,7 @@ from apps.wallet.sign_tx.scripts import ( if False: from typing import List + from trezor.crypto import bip32 # supported witness version for bech32 addresses _BECH32_WITVER = const(0x00) @@ -25,7 +27,10 @@ class AddressError(Exception): def get_address( - script_type: InputScriptType, coin: CoinInfo, node, multisig=None + script_type: int, + coin: CoinInfo, + node: bip32.HDNode, + multisig: MultisigRedeemScriptType = None, ) -> str: if ( @@ -88,7 +93,7 @@ def get_address( raise AddressError(FailureType.ProcessError, "Invalid script type") -def address_multisig_p2sh(pubkeys: List[bytes], m: int, coin: CoinInfo): +def address_multisig_p2sh(pubkeys: List[bytes], m: int, coin: CoinInfo) -> str: if coin.address_type_p2sh is None: raise AddressError( FailureType.ProcessError, "Multisig not enabled on this coin" @@ -98,7 +103,7 @@ def address_multisig_p2sh(pubkeys: List[bytes], m: int, coin: CoinInfo): return address_p2sh(redeem_script_hash, coin) -def address_multisig_p2wsh_in_p2sh(pubkeys: List[bytes], m: int, coin: CoinInfo): +def address_multisig_p2wsh_in_p2sh(pubkeys: List[bytes], m: int, coin: CoinInfo) -> str: if coin.address_type_p2sh is None: raise AddressError( FailureType.ProcessError, "Multisig not enabled on this coin" @@ -108,7 +113,7 @@ def address_multisig_p2wsh_in_p2sh(pubkeys: List[bytes], m: int, coin: CoinInfo) return address_p2wsh_in_p2sh(witness_script_hash, coin) -def address_multisig_p2wsh(pubkeys: List[bytes], m: int, hrp: str): +def address_multisig_p2wsh(pubkeys: List[bytes], m: int, hrp: str) -> str: if not hrp: raise AddressError( FailureType.ProcessError, "Multisig not enabled on this coin" @@ -196,7 +201,7 @@ def address_short(coin: CoinInfo, address: str) -> str: def validate_full_path( - path: list, coin: CoinInfo, script_type: InputScriptType, validate_script_type=True + path: list, coin: CoinInfo, script_type: int, validate_script_type: bool = True ) -> bool: """ Validates derivation path to fit Bitcoin-like coins. We mostly use @@ -244,9 +249,7 @@ def validate_purpose(purpose: int, coin: CoinInfo) -> bool: return True -def validate_purpose_against_script_type( - purpose: int, script_type: InputScriptType -) -> bool: +def validate_purpose_against_script_type(purpose: int, script_type: int) -> bool: """ Validates purpose against provided input's script type: - 44 for spending address (script_type == SPENDADDRESS) diff --git a/core/src/apps/wallet/sign_tx/bitcoinlike.py b/core/src/apps/wallet/sign_tx/bitcoinlike.py index 53402ac2e..8619b7736 100644 --- a/core/src/apps/wallet/sign_tx/bitcoinlike.py +++ b/core/src/apps/wallet/sign_tx/bitcoinlike.py @@ -44,7 +44,7 @@ class Bitcoinlike(signing.Bitcoin): else: await super().phase2_sign_nonsegwit_input(i_sign) - async def phase2_sign_bip143_input(self, i_sign) -> None: + async def phase2_sign_bip143_input(self, i_sign: int) -> None: # STAGE_REQUEST_SEGWIT_INPUT txi_sign = await helpers.request_tx_input(self.tx_req, i_sign, self.coin) self.input_check_wallet_path(txi_sign) diff --git a/core/src/apps/wallet/sign_tx/decred.py b/core/src/apps/wallet/sign_tx/decred.py index a295e58ef..6885c97b2 100644 --- a/core/src/apps/wallet/sign_tx/decred.py +++ b/core/src/apps/wallet/sign_tx/decred.py @@ -33,19 +33,19 @@ class DecredPrefixHasher: writers.write_uint32(self.h_prefix, tx.version | DECRED_SERIALIZE_NO_WITNESS) writers.write_varint(self.h_prefix, tx.inputs_count) - def add_prevouts(self, txi: TxInputType): + def add_prevouts(self, txi: TxInputType) -> None: writers.write_tx_input_decred(self.h_prefix, txi) - def add_sequence(self, txi: TxInputType): + def add_sequence(self, txi: TxInputType) -> None: pass - def add_output_count(self, tx: SignTx): + def add_output_count(self, tx: SignTx) -> None: writers.write_varint(self.h_prefix, tx.outputs_count) - def add_output(self, txo_bin: TxOutputBinType): + def add_output(self, txo_bin: TxOutputBinType) -> None: writers.write_tx_output(self.h_prefix, txo_bin) - def add_locktime_expiry(self, tx: SignTx): + def add_locktime_expiry(self, tx: SignTx) -> None: writers.write_uint32(self.h_prefix, tx.lock_time) writers.write_uint32(self.h_prefix, tx.expiry) @@ -54,21 +54,23 @@ class DecredPrefixHasher: class Decred(Bitcoin): - def initialize(self, tx: SignTx, keychain: seed.Keychain, coin: coininfo.CoinInfo): + def initialize( + self, tx: SignTx, keychain: seed.Keychain, coin: coininfo.CoinInfo + ) -> None: super().initialize(tx, keychain, coin) # This is required because the last serialized output obtained in # `check_fee` will only be sent to the client in `sign_tx` self.last_output_bytes = None # type: bytearray - def init_hash143(self): + def init_hash143(self) -> None: self.hash143 = DecredPrefixHasher(self.tx) # pseudo BIP-0143 prefix hashing - async def phase1(self): + async def phase1(self) -> None: await super().phase1() self.hash143.add_locktime_expiry(self.tx) - async def phase1_process_input(self, i: int, txi: TxInputType): + async def phase1_process_input(self, i: int, txi: TxInputType) -> None: await super().phase1_process_input(i, txi) w_txi = writers.empty_bytearray(8 if i == 0 else 0 + 9 + len(txi.prev_hash)) if i == 0: # serializing first input => prepend headers @@ -78,7 +80,7 @@ class Decred(Bitcoin): async def phase1_confirm_output( self, i: int, txo: TxOutputType, txo_bin: TxOutputBinType - ): + ) -> None: if txo.decred_script_version is not None and txo.decred_script_version != 0: raise SigningError( FailureType.ActionCancelled, @@ -97,7 +99,7 @@ class Decred(Bitcoin): await super().phase1_confirm_output(i, txo, txo_bin) - async def phase2(self): + async def phase2(self) -> None: self.tx_req.serialized = None prefix_hash = self.hash143.prefix_hash() @@ -171,7 +173,7 @@ class Decred(Bitcoin): i_sign, signature, w_txi_sign ) - return await helpers.request_tx_finish(self.tx_req) + await helpers.request_tx_finish(self.tx_req) async def get_prevtx_output_value(self, prev_hash: bytes, prev_index: int) -> int: total_out = 0 # sum of output amounts diff --git a/core/src/apps/wallet/sign_tx/helpers.py b/core/src/apps/wallet/sign_tx/helpers.py index 12eafb375..b154b5468 100644 --- a/core/src/apps/wallet/sign_tx/helpers.py +++ b/core/src/apps/wallet/sign_tx/helpers.py @@ -22,7 +22,7 @@ from .writers import TX_HASH_SIZE from apps.common.coininfo import CoinInfo if False: - from typing import Union + from typing import Any, Awaitable, Union MULTISIG_INPUT_SCRIPT_TYPES = ( InputScriptType.SPENDMULTISIG, @@ -92,27 +92,27 @@ class UiConfirmNonDefaultLocktime: __eq__ = utils.obj_eq -def confirm_output(output: TxOutputType, coin: CoinInfo): +def confirm_output(output: TxOutputType, coin: CoinInfo) -> Awaitable[Any]: # type: ignore return (yield UiConfirmOutput(output, coin)) -def confirm_total(spending: int, fee: int, coin: CoinInfo): +def confirm_total(spending: int, fee: int, coin: CoinInfo) -> Awaitable[Any]: # type: ignore return (yield UiConfirmTotal(spending, fee, coin)) -def confirm_feeoverthreshold(fee: int, coin: CoinInfo): +def confirm_feeoverthreshold(fee: int, coin: CoinInfo) -> Awaitable[Any]: # type: ignore return (yield UiConfirmFeeOverThreshold(fee, coin)) -def confirm_foreign_address(address_n: list): +def confirm_foreign_address(address_n: list) -> Awaitable[Any]: # type: ignore return (yield UiConfirmForeignAddress(address_n)) -def confirm_nondefault_locktime(lock_time: int): +def confirm_nondefault_locktime(lock_time: int) -> Awaitable[Any]: # type: ignore return (yield UiConfirmNonDefaultLocktime(lock_time)) -def request_tx_meta(tx_req: TxRequest, coin: CoinInfo, tx_hash: bytes = None): +def request_tx_meta(tx_req: TxRequest, coin: CoinInfo, tx_hash: bytes = None) -> Awaitable[Any]: # type: ignore tx_req.request_type = TXMETA tx_req.details.tx_hash = tx_hash tx_req.details.request_index = None @@ -122,9 +122,9 @@ def request_tx_meta(tx_req: TxRequest, coin: CoinInfo, tx_hash: bytes = None): return sanitize_tx_meta(ack.tx, coin) -def request_tx_extra_data( +def request_tx_extra_data( # type: ignore tx_req: TxRequest, offset: int, size: int, tx_hash: bytes = None -): +) -> Awaitable[Any]: tx_req.request_type = TXEXTRADATA tx_req.details.extra_data_offset = offset tx_req.details.extra_data_len = size @@ -138,7 +138,7 @@ def request_tx_extra_data( return ack.tx.extra_data -def request_tx_input(tx_req: TxRequest, i: int, coin: CoinInfo, tx_hash: bytes = None): +def request_tx_input(tx_req: TxRequest, i: int, coin: CoinInfo, tx_hash: bytes = None) -> Awaitable[Any]: # type: ignore tx_req.request_type = TXINPUT tx_req.details.request_index = i tx_req.details.tx_hash = tx_hash @@ -148,7 +148,7 @@ def request_tx_input(tx_req: TxRequest, i: int, coin: CoinInfo, tx_hash: bytes = return sanitize_tx_input(ack.tx, coin) -def request_tx_output(tx_req: TxRequest, i: int, coin: CoinInfo, tx_hash: bytes = None): +def request_tx_output(tx_req: TxRequest, i: int, coin: CoinInfo, tx_hash: bytes = None) -> Awaitable[Any]: # type: ignore tx_req.request_type = TXOUTPUT tx_req.details.request_index = i tx_req.details.tx_hash = tx_hash @@ -161,7 +161,7 @@ def request_tx_output(tx_req: TxRequest, i: int, coin: CoinInfo, tx_hash: bytes return sanitize_tx_binoutput(ack.tx, coin) -def request_tx_finish(tx_req: TxRequest): +def request_tx_finish(tx_req: TxRequest) -> Awaitable[Any]: # type: ignore tx_req.request_type = TXFINISHED tx_req.details = None yield tx_req @@ -293,7 +293,7 @@ def sanitize_tx_binoutput(tx: TransactionType, coin: CoinInfo) -> TxOutputBinTyp def _sanitize_decred( tx: Union[TxInputType, TxOutputType, TxOutputBinType], coin: CoinInfo -): +) -> None: if not coin.decred and tx.decred_script_version is not None: raise SigningError( FailureType.DataError, diff --git a/core/src/apps/wallet/sign_tx/layout.py b/core/src/apps/wallet/sign_tx/layout.py index 47c04f310..6974971ce 100644 --- a/core/src/apps/wallet/sign_tx/layout.py +++ b/core/src/apps/wallet/sign_tx/layout.py @@ -3,25 +3,34 @@ from ubinascii import hexlify from trezor import ui from trezor.messages import ButtonRequestType, OutputScriptType +from trezor.messages.TxOutputType import TxOutputType from trezor.strings import format_amount from trezor.utils import chunks +from apps.common import coininfo + +if False: + from typing import Iterator, List + from trezor import wire + _LOCKTIME_TIMESTAMP_MIN_VALUE = const(500000000) -def format_coin_amount(amount, coin): +def format_coin_amount(amount: int, coin: coininfo.CoinInfo) -> str: return "%s %s" % (format_amount(amount, coin.decimals), coin.coin_shortcut) -def split_address(address): +def split_address(address: str) -> Iterator[str]: return chunks(address, 17) -def split_op_return(data): +def split_op_return(data: str) -> Iterator[str]: return chunks(data, 18) -async def confirm_output(ctx, output, coin): +async def confirm_output( + ctx: wire.Context, output: TxOutputType, coin: coininfo.CoinInfo +) -> bool: from trezor.ui.text import Text from apps.common.confirm import confirm from apps.wallet.sign_tx import addresses, omni @@ -34,11 +43,11 @@ async def confirm_output(ctx, output, coin): text.normal(omni.parse(data)) else: # generic OP_RETURN - data = hexlify(data).decode() - if len(data) >= 18 * 5: - data = data[: (18 * 5 - 3)] + "..." + hex_data = hexlify(data).decode() + if len(hex_data) >= 18 * 5: + hex_data = hex_data[: (18 * 5 - 3)] + "..." text = Text("OP_RETURN", ui.ICON_SEND, ui.GREEN) - text.mono(*split_op_return(data)) + text.mono(*split_op_return(hex_data)) else: address = output.address address_short = addresses.address_short(coin, address) @@ -48,7 +57,9 @@ async def confirm_output(ctx, output, coin): return await confirm(ctx, text, ButtonRequestType.ConfirmOutput) -async def confirm_total(ctx, spending, fee, coin): +async def confirm_total( + ctx: wire.Context, spending: int, fee: int, coin: coininfo.CoinInfo +) -> bool: from trezor.ui.text import Text from apps.common.confirm import hold_to_confirm @@ -60,7 +71,9 @@ async def confirm_total(ctx, spending, fee, coin): return await hold_to_confirm(ctx, text, ButtonRequestType.SignTx) -async def confirm_feeoverthreshold(ctx, fee, coin): +async def confirm_feeoverthreshold( + ctx: wire.Context, fee: int, coin: coininfo.CoinInfo +) -> bool: from trezor.ui.text import Text from apps.common.confirm import confirm @@ -71,7 +84,9 @@ async def confirm_feeoverthreshold(ctx, fee, coin): return await confirm(ctx, text, ButtonRequestType.FeeOverThreshold) -async def confirm_foreign_address(ctx, address_n, coin): +async def confirm_foreign_address( + ctx: wire.Context, address_n: List[int], coin: coininfo.CoinInfo +) -> bool: from trezor.ui.text import Text from apps.common.confirm import confirm @@ -80,7 +95,7 @@ async def confirm_foreign_address(ctx, address_n, coin): return await confirm(ctx, text, ButtonRequestType.SignTx) -async def confirm_nondefault_locktime(ctx, lock_time): +async def confirm_nondefault_locktime(ctx: wire.Context, lock_time: int) -> bool: from trezor.ui.text import Text from apps.common.confirm import confirm diff --git a/core/src/apps/wallet/sign_tx/multisig.py b/core/src/apps/wallet/sign_tx/multisig.py index aff137dd6..d60df69fc 100644 --- a/core/src/apps/wallet/sign_tx/multisig.py +++ b/core/src/apps/wallet/sign_tx/multisig.py @@ -70,8 +70,8 @@ def multisig_fingerprint(multisig: MultisigRedeemScriptType) -> bytes: def multisig_pubkey_index(multisig: MultisigRedeemScriptType, pubkey: bytes) -> int: if multisig.nodes: - for i, hd in enumerate(multisig.nodes): - if multisig_get_pubkey(hd, multisig.address_n) == pubkey: + for i, hd_node in enumerate(multisig.nodes): + if multisig_get_pubkey(hd_node, multisig.address_n) == pubkey: return i else: for i, hd in enumerate(multisig.pubkeys): diff --git a/core/src/apps/wallet/sign_tx/scripts.py b/core/src/apps/wallet/sign_tx/scripts.py index d02171d0b..260ab5340 100644 --- a/core/src/apps/wallet/sign_tx/scripts.py +++ b/core/src/apps/wallet/sign_tx/scripts.py @@ -11,6 +11,10 @@ from apps.wallet.sign_tx.writers import ( write_varint, ) +if False: + from typing import List + from apps.wallet.sign_tx.writers import Writer + class ScriptsError(ValueError): pass @@ -145,7 +149,7 @@ def witness_p2wsh( signature: bytes, signature_index: int, sighash: int, -): +) -> bytearray: # get other signatures, stretch with None to the number of the pubkeys signatures = multisig.signatures + [None] * ( multisig_get_pubkey_count(multisig) - len(multisig.signatures) @@ -184,7 +188,7 @@ def witness_p2wsh( # redeem script write_varint(w, redeem_script_length) - output_script_multisig(pubkeys, multisig.m, w) + write_output_script_multisig(w, pubkeys, multisig.m) return w @@ -233,12 +237,18 @@ def input_script_multisig( # redeem script write_op_push(w, redeem_script_length) - output_script_multisig(pubkeys, multisig.m, w) + write_output_script_multisig(w, pubkeys, multisig.m) + + return w + +def output_script_multisig(pubkeys: List[bytes], m: int, w: Writer = None) -> bytearray: + w = empty_bytearray(output_script_multisig_length(pubkeys, m)) + write_output_script_multisig(w, pubkeys, m) return w -def output_script_multisig(pubkeys, m: int, w: bytearray = None) -> bytearray: +def write_output_script_multisig(w: Writer, pubkeys: List[bytes], m: int) -> None: n = len(pubkeys) if n < 1 or n > 15 or m < 1 or m > 15 or m > n: raise ScriptsError("Invalid multisig parameters") @@ -246,17 +256,14 @@ def output_script_multisig(pubkeys, m: int, w: bytearray = None) -> bytearray: if len(pubkey) != 33: raise ScriptsError("Invalid multisig parameters") - if w is None: - w = empty_bytearray(output_script_multisig_length(pubkeys, m)) w.append(0x50 + m) # numbers 1 to 16 are pushed as 0x50 + value for p in pubkeys: append_pubkey(w, p) w.append(0x50 + n) w.append(0xAE) # OP_CHECKMULTISIG - return w -def output_script_multisig_length(pubkeys, m: int) -> int: +def output_script_multisig_length(pubkeys: List[bytes], m: int) -> int: return 1 + len(pubkeys) * (1 + 33) + 1 + 1 # see output_script_multisig @@ -276,14 +283,12 @@ def output_script_paytoopreturn(data: bytes) -> bytearray: # === -def append_signature(w: bytearray, signature: bytes, sighash: int) -> bytearray: +def append_signature(w: Writer, signature: bytes, sighash: int) -> None: write_op_push(w, len(signature) + 1) write_bytes_unchecked(w, signature) w.append(sighash) - return w -def append_pubkey(w: bytearray, pubkey: bytes) -> bytearray: +def append_pubkey(w: Writer, pubkey: bytes) -> None: write_op_push(w, len(pubkey)) write_bytes_unchecked(w, pubkey) - return w diff --git a/core/src/apps/wallet/sign_tx/segwit_bip143.py b/core/src/apps/wallet/sign_tx/segwit_bip143.py index 40c26db7f..e5f07a4ab 100644 --- a/core/src/apps/wallet/sign_tx/segwit_bip143.py +++ b/core/src/apps/wallet/sign_tx/segwit_bip143.py @@ -25,19 +25,19 @@ class Bip143Error(ValueError): class Bip143: - def __init__(self): + def __init__(self) -> None: self.h_prevouts = HashWriter(sha256()) self.h_sequence = HashWriter(sha256()) self.h_outputs = HashWriter(sha256()) - def add_prevouts(self, txi: TxInputType): + def add_prevouts(self, txi: TxInputType) -> None: write_bytes_reversed(self.h_prevouts, txi.prev_hash, TX_HASH_SIZE) write_uint32(self.h_prevouts, txi.prev_index) - def add_sequence(self, txi: TxInputType): + def add_sequence(self, txi: TxInputType) -> None: write_uint32(self.h_sequence, txi.sequence) - def add_output(self, txo_bin: TxOutputBinType): + def add_output(self, txo_bin: TxOutputBinType) -> None: write_tx_output(self.h_outputs, txo_bin) def get_prevouts_hash(self, coin: CoinInfo) -> bytes: diff --git a/core/src/apps/wallet/sign_tx/signing.py b/core/src/apps/wallet/sign_tx/signing.py index ef4dc69f9..4f36b9421 100644 --- a/core/src/apps/wallet/sign_tx/signing.py +++ b/core/src/apps/wallet/sign_tx/signing.py @@ -246,7 +246,7 @@ class Bitcoin: await helpers.request_tx_finish(self.tx_req) - async def phase2_serialize_segwit_input(self, i_sign) -> None: + async def phase2_serialize_segwit_input(self, i_sign: int) -> None: # STAGE_REQUEST_SEGWIT_INPUT txi_sign = await helpers.request_tx_input(self.tx_req, i_sign, self.coin) @@ -270,7 +270,7 @@ class Bitcoin: writers.write_tx_input(w_txi, txi_sign) self.tx_req.serialized = TxRequestSerializedType(serialized_tx=w_txi) - async def phase2_sign_segwit_input(self, i) -> Tuple[bytearray, bytes]: + async def phase2_sign_segwit_input(self, i: int) -> Tuple[bytearray, bytes]: txi = await helpers.request_tx_input(self.tx_req, i, self.coin) self.input_check_wallet_path(txi) @@ -552,8 +552,8 @@ class Bitcoin: # p2wsh in p2sh pubkeys = multisig.multisig_get_pubkeys(i.multisig) witness_script_hasher = utils.HashWriter(sha256()) - scripts.output_script_multisig( - pubkeys, i.multisig.m, witness_script_hasher + scripts.write_output_script_multisig( + witness_script_hasher, pubkeys, i.multisig.m ) witness_script_hash = witness_script_hasher.get_digest() return scripts.input_script_p2wsh_in_p2sh(witness_script_hash) diff --git a/core/src/apps/wallet/sign_tx/tx_weight.py b/core/src/apps/wallet/sign_tx/tx_weight.py index ea0edf4bd..56eabf496 100644 --- a/core/src/apps/wallet/sign_tx/tx_weight.py +++ b/core/src/apps/wallet/sign_tx/tx_weight.py @@ -44,13 +44,13 @@ class TxWeightCalculator: ) self.segwit = False - def add_witness_header(self): + def add_witness_header(self) -> None: if not self.segwit: self.counter += _TXSIZE_SEGWIT_OVERHEAD self.counter += self.ser_length_size(self.inputs_count) self.segwit = True - def add_input(self, i: TxInputType): + def add_input(self, i: TxInputType) -> None: if i.multisig: multisig_script_size = _TXSIZE_MULTISIGSCRIPT + len(i.multisig.pubkeys) * ( @@ -88,7 +88,7 @@ class TxWeightCalculator: self.counter += 4 # empty self.counter += input_script_size # discounted witness - def add_output(self, script: bytes): + def add_output(self, script: bytes) -> None: size = len(script) + self.ser_length_size(len(script)) self.counter += 4 * (_TXSIZE_OUTPUT + size) @@ -96,7 +96,7 @@ class TxWeightCalculator: return self.counter @staticmethod - def ser_length_size(length: int): + def ser_length_size(length: int) -> int: if length < 253: return 1 if length < 0x10000: @@ -104,7 +104,7 @@ class TxWeightCalculator: return 5 @staticmethod - def op_push_size(length: int): + def op_push_size(length: int) -> int: if length < 0x4C: return 1 if length < 0x100: diff --git a/core/src/apps/wallet/sign_tx/writers.py b/core/src/apps/wallet/sign_tx/writers.py index 5482457cf..be4912994 100644 --- a/core/src/apps/wallet/sign_tx/writers.py +++ b/core/src/apps/wallet/sign_tx/writers.py @@ -18,6 +18,7 @@ from apps.common.writers import ( # noqa: F401 if False: from apps.common.writers import Writer + from trezor.utils import HashWriter write_uint16 = write_uint16_le write_uint32 = write_uint32_le @@ -31,14 +32,14 @@ def write_bytes_prefixed(w: Writer, b: bytes) -> None: write_bytes_unchecked(w, b) -def write_tx_input(w, i: TxInputType): +def write_tx_input(w: Writer, i: TxInputType) -> None: write_bytes_reversed(w, i.prev_hash, TX_HASH_SIZE) write_uint32(w, i.prev_index) write_bytes_prefixed(w, i.script_sig) write_uint32(w, i.sequence) -def write_tx_input_check(w, i: TxInputType): +def write_tx_input_check(w: Writer, i: TxInputType) -> None: write_bytes_fixed(w, i.prev_hash, TX_HASH_SIZE) write_uint32(w, i.prev_index) write_uint32(w, i.script_type) @@ -49,28 +50,28 @@ def write_tx_input_check(w, i: TxInputType): write_uint64(w, i.amount or 0) -def write_tx_input_decred(w, i: TxInputType): +def write_tx_input_decred(w: Writer, i: TxInputType) -> None: write_bytes_reversed(w, i.prev_hash, TX_HASH_SIZE) write_uint32(w, i.prev_index or 0) write_uint8(w, i.decred_tree or 0) write_uint32(w, i.sequence) -def write_tx_input_decred_witness(w, i: TxInputType): +def write_tx_input_decred_witness(w: Writer, i: TxInputType) -> None: write_uint64(w, i.amount or 0) write_uint32(w, 0) # block height fraud proof write_uint32(w, 0xFFFFFFFF) # block index fraud proof write_bytes_prefixed(w, i.script_sig) -def write_tx_output(w, o: TxOutputBinType): +def write_tx_output(w: Writer, o: TxOutputBinType) -> None: write_uint64(w, o.amount) if o.decred_script_version is not None: write_uint16(w, o.decred_script_version) write_bytes_prefixed(w, o.script_pubkey) -def write_op_push(w, n: int): +def write_op_push(w: Writer, n: int) -> None: ensure(n >= 0 and n <= 0xFFFFFFFF) if n < 0x4C: w.append(n & 0xFF) @@ -89,7 +90,7 @@ def write_op_push(w, n: int): w.append((n >> 24) & 0xFF) -def write_varint(w, n: int): +def write_varint(w: Writer, n: int) -> None: ensure(n >= 0 and n <= 0xFFFFFFFF) if n < 253: w.append(n & 0xFF) @@ -105,7 +106,7 @@ def write_varint(w, n: int): w.append((n >> 24) & 0xFF) -def get_tx_hash(w, double: bool = False, reverse: bool = False) -> bytes: +def get_tx_hash(w: HashWriter, double: bool = False, reverse: bool = False) -> bytes: d = w.get_digest() if double: d = sha256(d).digest() diff --git a/core/src/apps/wallet/sign_tx/zcash.py b/core/src/apps/wallet/sign_tx/zcash.py index cde31b5c4..f2eda8e2e 100644 --- a/core/src/apps/wallet/sign_tx/zcash.py +++ b/core/src/apps/wallet/sign_tx/zcash.py @@ -53,14 +53,14 @@ class Zip143: self.h_sequence = HashWriter(blake2b(outlen=32, personal=b"ZcashSequencHash")) self.h_outputs = HashWriter(blake2b(outlen=32, personal=b"ZcashOutputsHash")) - def add_prevouts(self, txi: TxInputType): + def add_prevouts(self, txi: TxInputType) -> None: write_bytes_reversed(self.h_prevouts, txi.prev_hash, TX_HASH_SIZE) write_uint32(self.h_prevouts, txi.prev_index) - def add_sequence(self, txi: TxInputType): + def add_sequence(self, txi: TxInputType) -> None: write_uint32(self.h_sequence, txi.sequence) - def add_output(self, txo_bin: TxOutputBinType): + def add_output(self, txo_bin: TxOutputBinType) -> None: write_tx_output(self.h_outputs, txo_bin) def get_prevouts_hash(self) -> bytes: @@ -119,7 +119,7 @@ class Zip143: class Zip243(Zip143): - def __init__(self, branch_id) -> None: + def __init__(self, branch_id: int) -> None: super().__init__(branch_id) def preimage_hash( diff --git a/core/src/trezor/crypto/bech32.py b/core/src/trezor/crypto/bech32.py index d7a82cf40..ec3f2c4d4 100644 --- a/core/src/trezor/crypto/bech32.py +++ b/core/src/trezor/crypto/bech32.py @@ -21,7 +21,7 @@ """Reference implementation for Bech32 and segwit addresses.""" if False: - from typing import List, Optional, Tuple + from typing import Iterable, List, Optional, Tuple CHARSET = "qpzry9x8gf2tvdw0s3jn54khce6mua7l" @@ -81,7 +81,7 @@ def bech32_decode(bech: str) -> Tuple[Optional[str], Optional[List[int]]]: def convertbits( - data: List[int], frombits: int, tobits: int, pad: bool = True + data: Iterable[int], frombits: int, tobits: int, pad: bool = True ) -> List[int]: """General power-of-2 base conversion.""" acc = 0 @@ -120,7 +120,7 @@ def decode(hrp: str, addr: str) -> Tuple[Optional[int], Optional[List[int]]]: return (data[0], decoded) -def encode(hrp: str, witver: int, witprog: List[int]) -> Optional[str]: +def encode(hrp: str, witver: int, witprog: Iterable[int]) -> Optional[str]: """Encode a segwit address.""" ret = bech32_encode(hrp, [witver] + convertbits(witprog, 8, 5)) if decode(hrp, ret) == (None, None): diff --git a/core/src/trezor/utils.py b/core/src/trezor/utils.py index 0232238b2..b21ac8bf8 100644 --- a/core/src/trezor/utils.py +++ b/core/src/trezor/utils.py @@ -86,9 +86,6 @@ if False: def extend(self, buf: bytes) -> None: ... - def write(self, buf: bytes) -> None: - ... - class HashWriter: def __init__(self, ctx: HashContext) -> None: