style(all): fix pylint complaints

[no changelog]
pull/1879/head
matejcik 3 years ago committed by matejcik
parent f88f568f1e
commit e972839ac7

@ -5,7 +5,7 @@ from os.path import basename
from graphviz import Digraph
class Message(object):
class Message:
def __init__(self, fname, name, attrs):
self.fname = basename(fname)
self.name = name

@ -64,6 +64,7 @@ def summary(coins, api_key):
except Exception:
pass
marketcap_percent = 100 * supported_marketcap / total_marketcap
return dict(
updated_at=int(time.time()),
updated_at_readable=time.asctime(),
@ -71,9 +72,7 @@ def summary(coins, api_key):
t2_coins=t2_coins,
marketcap_usd=supported_marketcap,
total_marketcap_usd=total_marketcap,
marketcap_supported="{:.02f} %".format(
100 * supported_marketcap / total_marketcap
),
marketcap_supported=f"{marketcap_percent:.02f} %",
)

@ -78,7 +78,7 @@ def c_str_filter(b):
return "NULL"
def hexescape(c):
return r"\x{:02x}".format(c)
return rf"\x{c:02x}"
if isinstance(b, bytes):
return '"' + "".join(map(hexescape, b)) + '"'

@ -128,13 +128,12 @@ def write_bip143_script_code_prefixed(
write_output_script_multisig(w, public_keys, threshold, prefixed=True)
return
p2pkh = (
txi.script_type == InputScriptType.SPENDWITNESS
or txi.script_type == InputScriptType.SPENDP2SHWITNESS
or txi.script_type == InputScriptType.SPENDADDRESS
or txi.script_type == InputScriptType.EXTERNAL
p2pkh = txi.script_type in (
InputScriptType.SPENDWITNESS,
InputScriptType.SPENDP2SHWITNESS,
InputScriptType.SPENDADDRESS,
InputScriptType.EXTERNAL,
)
if p2pkh:
# for p2wpkh in p2sh or native p2wpkh
# the scriptCode is a classic p2pkh
@ -367,7 +366,7 @@ def parse_witness_multisig(
raise ValueError
signatures = []
for i in range(item_count - 2):
for _ in range(item_count - 2):
n = read_bitcoin_varint(r)
signature = r.read_memoryview(n - 1)
hash_type = r.get()
@ -502,7 +501,7 @@ def parse_output_script_multisig(script: bytes) -> tuple[list[memoryview], int]:
raise ValueError
public_keys = []
for i in range(pubkey_count):
for _ in range(pubkey_count):
n = read_op_push(r)
if n != 33:
raise ValueError

@ -40,7 +40,7 @@ def write_input_script_prefixed(
# p2sh multisig
assert multisig is not None # checked in sanitize_tx_input
signature_index = multisig_pubkey_index(multisig, pubkey)
return write_input_script_multisig_prefixed(
write_input_script_multisig_prefixed(
w, multisig, signature, signature_index, hash_type, coin
)
else:

@ -37,7 +37,7 @@ if False:
]
class SignerClass(Protocol):
def __init__(
def __init__( # pylint: disable=super-init-not-called
self,
tx: SignTx,
keychain: Keychain,

@ -9,6 +9,8 @@ from micropython import const
from trezor.enums import InputScriptType
from .. import common
if False:
from trezor.messages import TxInput
@ -69,17 +71,11 @@ class TxWeightCalculator:
self.counter += 4 * _TXSIZE_INPUT
if (
i.script_type == InputScriptType.SPENDADDRESS
or i.script_type == InputScriptType.SPENDMULTISIG
):
if i.script_type in common.NONSEGWIT_INPUT_SCRIPT_TYPES:
input_script_size += self.ser_length_size(input_script_size)
self.counter += 4 * input_script_size
elif (
i.script_type == InputScriptType.SPENDWITNESS
or i.script_type == InputScriptType.SPENDP2SHWITNESS
):
elif i.script_type in common.SEGWIT_INPUT_SCRIPT_TYPES:
self.add_witness_header()
if i.script_type == InputScriptType.SPENDP2SHWITNESS:
if i.multisig:

@ -29,14 +29,14 @@ async def verify_message(ctx: wire.Context, msg: VerifyMessage) -> Success:
digest = message_digest(coin, message)
recid = signature[0]
if recid >= 27 and recid <= 34:
if 27 <= recid <= 34:
# p2pkh
script_type = InputScriptType.SPENDADDRESS
elif recid >= 35 and recid <= 38:
elif 35 <= recid <= 38:
# segwit-in-p2sh
script_type = InputScriptType.SPENDP2SHWITNESS
signature = bytes([signature[0] - 4]) + signature[1:]
elif recid >= 39 and recid <= 42:
elif 39 <= recid <= 42:
# native segwit
script_type = InputScriptType.SPENDWITNESS
signature = bytes([signature[0] - 8]) + signature[1:]

@ -61,7 +61,7 @@ def write_tx_output(w: Writer, o: TxOutput | PrevOutput, script_pubkey: bytes) -
def write_op_push(w: Writer, n: int) -> None:
ensure(n >= 0 and n <= 0xFFFF_FFFF)
ensure(0 <= n <= 0xFFFF_FFFF)
if n < 0x4C:
w.append(n & 0xFF)
elif n < 0xFF:
@ -80,7 +80,7 @@ def write_op_push(w: Writer, n: int) -> None:
def op_push_length(n: int) -> int:
ensure(n >= 0 and n <= 0xFFFF_FFFF)
ensure(0 <= n <= 0xFFFF_FFFF)
if n < 0x4C:
return 1
elif n < 0xFF:

@ -294,7 +294,7 @@ def _validate_shelley_address(
def _validate_address_size(address_bytes: bytes) -> None:
if not (MIN_ADDRESS_BYTES_LENGTH <= len(address_bytes) <= MAX_ADDRESS_BYTES_LENGTH):
if not MIN_ADDRESS_BYTES_LENGTH <= len(address_bytes) <= MAX_ADDRESS_BYTES_LENGTH:
raise INVALID_ADDRESS

@ -93,9 +93,13 @@ def is_printable_ascii_bytestring(bytestr: bytes) -> bool:
async def show_native_script(
ctx: wire.Context,
script: CardanoNativeScript,
indices: list[int] = [],
indices: list[int] | None = None,
) -> None:
indices_str = ".".join([str(i) for i in indices])
script_heading = "Script"
if indices is None:
indices = []
if indices:
script_heading += " " + ".".join(str(i) for i in indices)
script_type_name_suffix = ""
if script.type == CardanoNativeScriptType.PUB_KEY:
@ -104,11 +108,6 @@ async def show_native_script(
elif script.key_hash:
script_type_name_suffix = "hash"
if indices_str:
script_heading = "Script " + indices_str
else:
script_heading = "Script"
props: list[PropertyType] = [
(
f"{script_heading} - {SCRIPT_TYPE_NAMES[script.type]} {script_type_name_suffix}:",
@ -154,7 +153,7 @@ async def show_native_script(
)
for i, sub_script in enumerate(script.scripts):
await show_native_script(ctx, sub_script, indices + [(i + 1)])
await show_native_script(ctx, sub_script, indices + [i + 1])
async def show_script_hash(
@ -433,11 +432,9 @@ async def confirm_stake_pool_parameters(
),
("Pool reward account:", pool_parameters.reward_account),
(
"Pledge: {}\nCost: {}\nMargin: {}%".format(
format_coin_amount(pool_parameters.pledge),
format_coin_amount(pool_parameters.cost),
percentage_formatted,
),
f"Pledge: {format_coin_amount(pool_parameters.pledge)}\n"
+ f"Cost: {format_coin_amount(pool_parameters.cost)}\n"
+ f"Margin: {percentage_formatted}%",
None,
),
],

@ -313,7 +313,7 @@ async def _process_inputs(
inputs_count: int,
) -> None:
"""Read, validate and serialize the inputs."""
for index in range(inputs_count):
for _ in range(inputs_count):
input: CardanoTxInput = await ctx.call(CardanoTxItemAck(), CardanoTxInput)
inputs_list.append((input.prev_hash, input.prev_index))

@ -77,7 +77,7 @@ def _cbor_encode(value: Value) -> Iterator[bytes]:
encoded_value = value.encode()
yield _header(_CBOR_TEXT_STRING, len(encoded_value))
yield encoded_value
elif isinstance(value, list) or isinstance(value, tuple):
elif isinstance(value, (list, tuple)):
# definite-length valued list
yield _header(_CBOR_ARRAY, len(value))
for x in value:

@ -256,11 +256,8 @@ class PathSchema:
for component in self.schema:
if isinstance(component, Interval):
a, b = component.min, component.max
components.append(
"[{}-{}]{}".format(
unharden(a), unharden(b), "'" if a & HARDENED else ""
)
)
prime = "'" if a & HARDENED else ""
components.append(f"[{unharden(a)}-{unharden(b)}]{prime}")
else:
# mypy thinks component is a Contanier but we're using it as a Collection.
# Which in practice it is, the only non-Collection is Interval.

@ -87,7 +87,7 @@ def write_bytes_reversed(w: Writer, b: bytes, length: int) -> int:
def write_bitcoin_varint(w: Writer, n: int) -> None:
ensure(n >= 0 and n <= 0xFFFF_FFFF)
ensure(0 <= n <= 0xFFFF_FFFF)
if n < 253:
w.append(n & 0xFF)
elif n < 0x1_0000:
@ -103,7 +103,7 @@ def write_bitcoin_varint(w: Writer, n: int) -> None:
def write_uvarint(w: Writer, n: int) -> None:
ensure(n >= 0 and n <= 0xFFFF_FFFF_FFFF_FFFF)
ensure(0 <= n <= 0xFFFF_FFFF_FFFF_FFFF)
shifted = 1
while shifted:
shifted = n >> 7

@ -99,7 +99,7 @@ async def process_unknown_action(
def check_action(action: EosTxActionAck, name: str, account: str) -> bool:
if account == "eosio":
if (
return (
(name == "buyram" and action.buy_ram is not None)
or (name == "buyrambytes" and action.buy_ram_bytes is not None)
or (name == "sellram" and action.sell_ram is not None)
@ -112,10 +112,7 @@ def check_action(action: EosTxActionAck, name: str, account: str) -> bool:
or (name == "linkauth" and action.link_auth is not None)
or (name == "unlinkauth" and action.unlink_auth is not None)
or (name == "newaccount" and action.new_account is not None)
):
return True
else:
return False
)
elif name == "transfer":
return action.transfer is not None

@ -28,7 +28,7 @@ def eos_asset_to_string(asset: EosAsset) -> str:
precision = symbol_bytes[7]
symbol = bytes(reversed(symbol_bytes[:7])).rstrip(b"\x00").decode("ascii")
amount_digits = "{:0{precision}d}".format(asset.amount, precision=precision)
amount_digits = f"{asset.amount:0{precision}d}"
if precision > 0:
integer = amount_digits[:-precision]
if integer == "":

@ -49,6 +49,6 @@ async def _init(ctx: wire.Context, sha: HashWriter, msg: EosSignTx) -> None:
async def _actions(ctx: wire.Context, sha: HashWriter, num_actions: int) -> None:
for i in range(num_actions):
for _ in range(num_actions):
action = await ctx.call(EosTxActionRequest(), EosTxActionAck)
await process_action(ctx, sha, action)

@ -80,7 +80,7 @@ def _schemas_from_chain_id(msg: EthereumSignTxAny) -> Iterable[paths.PathSchema]
if info is None:
# allow Ethereum or testnet paths for unknown networks
slip44_id = (60, 1)
elif info.slip44 != 60 and info.slip44 != 1:
elif info.slip44 not in (60, 1):
# allow cross-signing with Ethereum unless it's testnet
slip44_id = (info.slip44, 60)
else:

@ -40,4 +40,4 @@ class HomescreenBase(ui.Layout):
def _before_render(self) -> None:
if storage.cache.homescreen_shown is not self.RENDER_INDICATOR:
return super()._before_render()
super()._before_render()

@ -224,8 +224,8 @@ async def _show_remaining_groups_and_shares(ctx: wire.GenericContext) -> None:
groups = set()
first_entered_index = -1
for i in range(len(shares_remaining)):
if shares_remaining[i] < slip39.MAX_SHARE_COUNT:
for i, group_count in enumerate(shares_remaining):
if group_count < slip39.MAX_SHARE_COUNT:
first_entered_index = i
share = None

@ -46,7 +46,7 @@ def process_slip39(words: str) -> tuple[bytes | None, slip39.Share]:
# if share threshold and group threshold are 1
# we can calculate the secret right away
if share.threshold == 1 and share.group_threshold == 1:
identifier, iteration_exponent, secret = slip39.recover_ems([words])
_, _, secret = slip39.recover_ems([words])
return secret, share
else:
# we need more shares
@ -87,7 +87,7 @@ def process_slip39(words: str) -> tuple[bytes | None, slip39.Share]:
# in case of slip39 basic we only need the first and only group
mnemonics = storage.recovery_shares.fetch_group(0)
identifier, iteration_exponent, secret = slip39.recover_ems(mnemonics)
_, _, secret = slip39.recover_ems(mnemonics)
return secret, share

@ -124,8 +124,6 @@ def sign_challenge(
sigtype: str | coininfo.CoinInfo,
curve: str,
) -> bytes:
from trezor.crypto.hashlib import sha256
if curve == "secp256k1":
from trezor.crypto.curve import secp256k1
elif curve == "nist256p1":
@ -166,7 +164,7 @@ def sign_challenge(
if curve == "ed25519":
signature = b"\x00" + signature
elif sigtype == "gpg" or sigtype == "ssh":
elif sigtype in ("gpg", "ssh"):
signature = b"\x00" + signature[1:]
return signature

@ -246,12 +246,10 @@ def _check_change(state: State, outputs: list[MoneroTransactionDestinationEntry]
state.mem_trace("No change" if __debug__ else None)
return
"""
Sweep tx is just one output and no change.
To prevent recognition of such transactions another fake output is added
that spends exactly 0 coins to a random address.
See https://github.com/monero-project/monero/pull/1415
"""
# Sweep tx is just one output and no change.
# To prevent recognition of such transactions another fake output is added
# that spends exactly 0 coins to a random address.
# See https://github.com/monero-project/monero/pull/1415
if change_index is None and state.output_change.amount == 0 and len(outputs) == 2:
state.mem_trace("Sweep tsx" if __debug__ else None)
return

@ -52,11 +52,9 @@ async def set_input(
tx_key = crypto.decodepoint(src_entr.real_out_tx_key)
additional_tx_pub_key = _get_additional_public_key(src_entr)
"""
Calculates `derivation = Ra`, private spend key `x = H(Ra||i) + b` to be able
to spend the UTXO; and key image `I = x*H(P||i)`
"""
xi, ki, di = monero.generate_tx_spend_and_key_image_and_derivation(
# Calculates `derivation = Ra`, private spend key `x = H(Ra||i) + b` to be able
# to spend the UTXO; and key image `I = x*H(P||i)`
xi, ki, _di = monero.generate_tx_spend_and_key_image_and_derivation(
state.creds,
state.subaddresses,
out_key,
@ -78,11 +76,9 @@ async def set_input(
if src_entr.rct:
vini.amount = 0
"""
Serialize `vini` with variant code for TxinToKey (prefix = TxinToKey.VARIANT_CODE).
The binary `vini_bin` is later sent to step 4 and 9 with its hmac,
where it is checked and directly used.
"""
# Serialize `vini` with variant code for TxinToKey (prefix = TxinToKey.VARIANT_CODE).
# The binary `vini_bin` is later sent to step 4 and 9 with its hmac,
# where it is checked and directly used.
vini_bin = serialize.dump_msg(vini, preallocate=64, prefix=b"\x02")
state.mem_trace(2, True)
@ -114,10 +110,8 @@ async def set_input(
state.last_step = state.STEP_INP
if state.current_input_index + 1 == state.input_count:
"""
When we finish the inputs processing, we no longer need
the precomputed subaddresses so we clear them to save memory.
"""
# When we finish the inputs processing, we no longer need
# the precomputed subaddresses so we clear them to save memory.
state.subaddresses = None
state.input_last_amount = src_entr.amount

@ -27,13 +27,13 @@ if False:
async def tsx_inputs_permutation(
state: State, permutation: list[int]
) -> MoneroTransactionInputsPermutationAck:
"""
Set permutation on the inputs - sorted by key image on host.
"""
from trezor.messages import MoneroTransactionInputsPermutationAck
await transaction_step(state, state.STEP_PERM)
"""
Set permutation on the inputs - sorted by key image on host.
"""
if state.last_step != state.STEP_INP:
raise ValueError("Invalid state transition")
if len(permutation) != state.input_count:

@ -54,9 +54,7 @@ async def input_vini(
if state.current_input_index > 0 and state.last_ki <= cur_ki:
raise ValueError("Key image order invalid")
"""
Incremental hasing of tx.vin[i]
"""
# Incremental hasing of tx.vin[i]
state.tx_prefix_hasher.buffer(vini_bin)
state.last_step = state.STEP_VINI
state.last_ki = cur_ki if state.current_input_index < state.input_count else None

@ -266,15 +266,15 @@ def _range_proof(
state.mem_trace("pre-rproof" if __debug__ else None, collect=True)
if not state.rsig_offload:
"""Bulletproof calculation in Trezor"""
# Bulletproof calculation in Trezor
rsig = _rsig_bp(state)
elif not state.is_processing_offloaded:
"""Bulletproof offloaded to the host, deterministic masks. Nothing here, waiting for offloaded BP."""
# Bulletproof offloaded to the host, deterministic masks. Nothing here, waiting for offloaded BP.
pass
else:
"""Bulletproof offloaded to the host, check BP, hash it."""
# Bulletproof offloaded to the host, check BP, hash it.
_rsig_process_bp(state, rsig_data)
state.mem_trace("rproof" if __debug__ else None, collect=True)

@ -239,11 +239,9 @@ def _protect_signature(state: State, mg_buffer: list[bytes]) -> list[bytes]:
cipher = chacha20poly1305(key, nonce)
"""
cipher.update() input has to be 512 bit long (besides the last block).
Thus we go over mg_buffer and buffer 512 bit input blocks before
calling cipher.update().
"""
# cipher.update() input has to be 512 bit long (besides the last block).
# Thus we go over mg_buffer and buffer 512 bit input blocks before
# calling cipher.update().
CHACHA_BLOCK = 64 # 512 bit chacha key-stream block size
buff = bytearray(CHACHA_BLOCK)
buff_len = 0 # valid bytes in the block buffer

@ -409,7 +409,7 @@ class KeyV(KeyVBase):
def resize(self, nsize, chop=False, realloc=False):
if self.size == nsize:
return self
return
if self.chunked and nsize <= _CHSIZE:
self.chunked = False # de-chunk

@ -215,13 +215,11 @@ def hash_to_scalar_into(r: Sc25519, data: bytes, length: int | None = None):
return tcry.xmr_hash_to_scalar(r, dt)
"""
H_p(buf)
Code adapted from MiniNero: https://github.com/monero-project/mininero
https://github.com/monero-project/research-lab/blob/master/whitepaper/ge_fromfe_writeup/ge_fromfe.pdf
http://archive.is/yfINb
"""
# H_p(buf)
#
# Code adapted from MiniNero: https://github.com/monero-project/mininero
# https://github.com/monero-project/research-lab/blob/master/whitepaper/ge_fromfe_writeup/ge_fromfe.pdf
# http://archive.is/yfINb
hash_to_point = tcry.xmr_hash_to_ec
hash_to_point_into = tcry.xmr_hash_to_ec

@ -75,7 +75,7 @@ def _export_key_image(
sub_addr_major,
sub_addr_minor,
)
xi, ki, recv_derivation = r[:3]
xi, ki, _ = r[:3]
phash = crypto.encodepoint(ki)
sig = generate_ring_signature(phash, ki, [pkey], xi, 0, test)

@ -33,13 +33,13 @@ class PreMlsagHasher:
self.rtcsig_hasher.uvarint(fee) # UVarintType
def set_ecdh(self, ecdh: bytes):
if self.state != 2 and self.state != 3 and self.state != 4:
if self.state not in (2, 3, 4):
raise ValueError("State error")
self.state = 4
self.rtcsig_hasher.buffer(ecdh)
def set_out_pk_commitment(self, out_pk_commitment: bytes):
if self.state != 4 and self.state != 5:
if self.state not in (4, 5):
raise ValueError("State error")
self.state = 5
self.rtcsig_hasher.buffer(out_pk_commitment) # ECKey

@ -134,7 +134,7 @@ class MessageType(XmrType):
"""
def __init__(self, **kwargs):
for kw in kwargs:
for kw in kwargs: # pylint: disable=consider-using-dict-items
setattr(self, kw, kwargs[kw])
__eq__ = obj_eq

@ -39,6 +39,8 @@ def get_network_str(network: int) -> str:
elif network == NEM_NETWORK_MIJIN:
return "Mijin"
raise RuntimeError # network should be one of the NEM_NETWORK_* constants
def check_path(path: paths.Bip32Path, network: int) -> bool:
"""Validates that the appropriate coin_type is set for the given network."""

@ -33,8 +33,9 @@ def address_from_public_key(pubkey: bytes) -> str:
- checksum is first 4 bytes of double sha256(data)
see https://developers.ripple.com/accounts.html#address-encoding
Returns the Ripple address created using base58
"""
"""Returns the Ripple address created using base58"""
h = sha256(pubkey).digest()
h = ripemd160(h).digest()

@ -262,12 +262,14 @@ def _format_thresholds(op: StellarSetOptionsOp) -> list[tuple[str, str]]:
def _format_flags(flags: int) -> str:
if flags > consts.FLAGS_MAX_SIZE:
raise ProcessError("Stellar: invalid flags")
text = "{}{}{}".format(
"AUTH_REQUIRED\n" if flags & consts.FLAG_AUTH_REQUIRED else "",
"AUTH_REVOCABLE\n" if flags & consts.FLAG_AUTH_REVOCABLE else "",
"AUTH_IMMUTABLE\n" if flags & consts.FLAG_AUTH_IMMUTABLE else "",
)
return text
flags_set = []
if flags & consts.FLAG_AUTH_REQUIRED:
flags_set.append("AUTH_REQUIRED\n")
if flags & consts.FLAG_AUTH_REVOCABLE:
flags_set.append("AUTH_REVOCABLE\n")
if flags & consts.FLAG_AUTH_IMMUTABLE:
flags_set.append("AUTH_IMMUTABLE\n")
return "".join(flags_set)
async def confirm_asset_issuer(ctx: Context, asset: StellarAsset) -> None:

@ -82,7 +82,7 @@ async def _timebounds(ctx: Context, w: bytearray, start: int, end: int) -> None:
async def _operations(ctx: Context, w: bytearray, num_operations: int) -> None:
writers.write_uint32(w, num_operations)
for i in range(num_operations):
for _ in range(num_operations):
op = await ctx.call_any(StellarTxOpRequest(), *consts.op_wire_types)
await process_operation(ctx, w, op) # type: ignore

@ -157,6 +157,8 @@ def _get_ballot(ballot):
elif ballot == TezosBallotType.Pass:
return "pass"
raise RuntimeError # unrecognized enum value
def _get_operation_bytes(w: bytearray, msg):
write_bytes_fixed(w, msg.branch, helpers.BRANCH_HASH_SIZE)

@ -43,7 +43,7 @@ def find_by_rp_id_hash(rp_id_hash: bytes) -> Iterator[Fido2Credential]:
def get_resident_credential(index: int) -> Fido2Credential | None:
if not (0 <= index < MAX_RESIDENT_CREDENTIALS):
if not 0 <= index < MAX_RESIDENT_CREDENTIALS:
return None
data = storage.resident_credentials.get(index)

@ -8,21 +8,21 @@ MAX_RESIDENT_CREDENTIALS = const(100)
def get(index: int) -> bytes | None:
if not (0 <= index < MAX_RESIDENT_CREDENTIALS):
if not 0 <= index < MAX_RESIDENT_CREDENTIALS:
raise ValueError # invalid credential index
return common.get(common.APP_WEBAUTHN, index + _RESIDENT_CREDENTIAL_START_KEY)
def set(index: int, data: bytes) -> None:
if not (0 <= index < MAX_RESIDENT_CREDENTIALS):
if not 0 <= index < MAX_RESIDENT_CREDENTIALS:
raise ValueError # invalid credential index
common.set(common.APP_WEBAUTHN, index + _RESIDENT_CREDENTIAL_START_KEY, data)
def delete(index: int) -> None:
if not (0 <= index < MAX_RESIDENT_CREDENTIALS):
if not 0 <= index < MAX_RESIDENT_CREDENTIALS:
raise ValueError # invalid credential index
common.delete(common.APP_WEBAUTHN, index + _RESIDENT_CREDENTIAL_START_KEY)

@ -35,7 +35,8 @@ def _get_device_dir() -> str:
def _get_salt_path(new: bool = False) -> str:
return "{}/salt{}".format(_get_device_dir(), ".new" if new else "")
ext = ".new" if new else ""
return f"{_get_device_dir()}/salt{ext}"
@with_filesystem

@ -8,7 +8,7 @@ from ustruct import unpack
_b32alphabet = "ABCDEFGHIJKLMNOPQRSTUVWXYZ234567"
_b32tab = [ord(c) for c in _b32alphabet]
_b32rev = dict([(ord(v), k) for k, v in enumerate(_b32alphabet)])
_b32rev = {ord(v): k for k, v in enumerate(_b32alphabet)}
def encode(s: bytes) -> str:
@ -54,7 +54,7 @@ def encode(s: bytes) -> str:
def decode(s: str) -> bytes:
data = s.encode()
quanta, leftover = divmod(len(data), 8)
_, leftover = divmod(len(data), 8)
if leftover:
raise ValueError("Incorrect padding")
# Strip off pad characters from the right. We need to count the pad
@ -78,11 +78,11 @@ def decode(s: str) -> bytes:
acc += _b32rev[c] << shift
shift -= 5
if shift < 0:
parts.append(unhexlify(("%010x" % acc).encode()))
parts.append(unhexlify((f"{acc:010x}").encode()))
acc = 0
shift = 35
# Process the last, partial quanta
last = unhexlify(bytes("%010x" % acc, "ascii"))
last = unhexlify(bytes(f"{acc:010x}", "ascii"))
if padchars == 0:
last = b"" # No characters
elif padchars == 1:

@ -51,7 +51,7 @@ def prefix_expand(prefix: str) -> list[int]:
def calculate_checksum(prefix: str, payload: list[int]) -> list[int]:
poly = cashaddr_polymod(prefix_expand(prefix) + payload + [0, 0, 0, 0, 0, 0, 0, 0])
out = list()
out = []
for i in range(8):
out.append((poly >> 5 * (7 - i)) & 0x1F)
return out
@ -62,7 +62,7 @@ def verify_checksum(prefix: str, payload: list[int]) -> bool:
def b32decode(inputs: str) -> list[int]:
out = list()
out = []
for letter in inputs:
out.append(CHARSET.find(letter))
return out

@ -20,8 +20,8 @@ def _byte_size(x: int) -> int:
for exp in range(64):
if x < 0x100 ** exp:
return exp
else:
raise ValueError # int is too large
raise ValueError # int is too large
def int_to_bytes(x: int) -> bytes:

@ -42,9 +42,8 @@ if False:
Indices = Tuple[int, ...]
MnemonicGroups = dict[int, tuple[int, set[tuple[int, bytes]]]]
"""
## Simple helpers
"""
# === Simple helpers ===
_RADIX_BITS = const(10)
"""The length of the radix in bits."""
@ -62,9 +61,7 @@ def _xor(a: bytes, b: bytes) -> bytes:
return bytes(x ^ y for x, y in zip(a, b))
"""
## Constants
"""
# === Constants ===
_ID_LENGTH_BITS = const(15)
"""The length of the random identifier in bits."""
@ -106,9 +103,7 @@ _DIGEST_INDEX = const(254)
"""The index of the share containing the digest of the shared secret."""
"""
# Keyboard functions
"""
# === Keyboard functions ===
KEYBOARD_FULL_MASK = const(0x1FF)
"""All buttons are allowed. 9-bit bitmap all set to 1."""
@ -126,9 +121,7 @@ def button_sequence_to_word(prefix: str) -> str:
return slip39.button_sequence_to_word(int(prefix))
"""
# External API
"""
# === External API ===
MAX_SHARE_COUNT = const(16)
"""The maximum number of shares that can be created."""
@ -265,7 +258,7 @@ def recover_ems(mnemonics: list[str]) -> tuple[int, int, bytes]:
identifier,
iteration_exponent,
group_threshold,
group_count,
_group_count,
groups,
) = _decode_mnemonics(mnemonics)
@ -274,7 +267,7 @@ def recover_ems(mnemonics: list[str]) -> tuple[int, int, bytes]:
f"Wrong number of mnemonic groups. Expected {group_threshold} groups, but {len(groups)} were provided."
)
for group_index, group in groups.items():
for group in groups.values():
if len(group[1]) != group[0]: # group[0] is threshold
raise MnemonicError(
f"Wrong number of mnemonics. Expected {group[0]} mnemonics, but {len(group[1])} were provided."
@ -344,9 +337,7 @@ def decode_mnemonic(mnemonic: str) -> Share:
)
"""
## Convert mnemonics or integers to indices and back
"""
# === Convert mnemonics or integers to indices and back ===
def _int_from_indices(indices: Indices) -> int:
@ -371,9 +362,7 @@ def _mnemonic_to_indices(mnemonic: str) -> Iterable[int]:
return (slip39.word_index(word.lower()) for word in mnemonic.split())
"""
## Checksum functions
"""
# === Checksum functions ===
def _rs1024_create_checksum(data: Indices) -> Indices:
@ -449,9 +438,7 @@ def _rs1024_error_index(data: Indices) -> int | None:
return None
"""
## Internal functions
"""
# === Internal functions ===
def _round_function(i: int, passphrase: bytes, e: int, salt: bytes, r: bytes) -> bytes:
@ -616,7 +603,7 @@ def _decode_mnemonics(
"Invalid set of mnemonics. All mnemonics must have the same group count."
)
for group_index, group in groups.items():
for group in groups.values():
if len(set(share[0] for share in group[1])) != len(group[1]):
raise MnemonicError(
"Invalid set of shares. Member indices in each group must be unique."

@ -99,7 +99,7 @@ def close(task: Task) -> None:
Unschedule and unblock a task, close it so it can release all resources, and
call its finalizer.
"""
for iface in _paused:
for iface in _paused: # pylint: disable=consider-using-dict-items
_paused[iface].discard(task)
_queue.discard(task)
task.close()

@ -1,3 +1,4 @@
# pylint: disable=wrong-import-position
import math
import utime
from micropython import const
@ -156,7 +157,7 @@ def header_error(message: str, clear: bool = True) -> None:
display.bar(0, 30, WIDTH, HEIGHT - 30, style.BG)
def draw_simple(t: Component) -> None: # noqa: F405
def draw_simple(t: "Component") -> None:
"""Render a component synchronously.
Useful when you need to put something on screen and go on to do other things.
@ -312,8 +313,6 @@ class Cancelled(Exception):
See `Layout.__iter__` for details.
"""
pass
class Layout(Component):
"""
@ -445,6 +444,7 @@ if utils.MODEL == "1":
class RustLayout(Layout):
def __init__(self, layout: Any):
super().__init__()
self.layout = layout
self.layout.set_timer_fn(self.set_timer)

@ -44,6 +44,7 @@ class ConfirmBase(ui.Layout):
confirm: ui.Component | None = None,
cancel: ui.Component | None = None,
) -> None:
super().__init__()
self.content = content
self.confirm = confirm
self.cancel = cancel

@ -274,7 +274,7 @@ def render_text(
# render it after a linebreak
item_width = ui.display.text_width(item, font)
if (
item_width <= line_width
item_width <= line_width # pylint: disable=chained-comparison
and item_width + offset_x - INITIAL_OFFSET_X > line_width
and "\n" not in item
):
@ -422,7 +422,7 @@ class TextBase(ui.Component):
param_font: int = ui.BOLD,
) -> None:
parts = format_string.split("{}", len(params))
for i in range(len(parts)):
for i in range(len(parts)): # pylint: disable=consider-using-enumerate
self.content.append(font)
self.content.append(parts[i])
if i < len(parts) - 1 and i < len(params):

@ -195,7 +195,7 @@ class Slip39Keyboard(ui.Layout):
def is_input_final(self) -> bool:
# returns True if mask has exactly one bit set to 1 or is 0
return not (self.mask & (self.mask - 1))
return not self.mask & (self.mask - 1)
def check_mask(self, index: int) -> bool:
return bool((1 << (index - 1)) & self.mask)

@ -108,6 +108,7 @@ class PinDialog(ui.Layout):
allow_cancel: bool = True,
maxlength: int = 50,
) -> None:
super().__init__()
self.maxlength = maxlength
self.input = PinInput(prompt, subprompt, "")

@ -310,7 +310,7 @@ async def show_address(
title: str = "Confirm address",
network: str | None = None,
multisig_index: int | None = None,
xpubs: Sequence[str] = [],
xpubs: Sequence[str] = (),
address_extra: str | None = None,
title_qr: str | None = None,
) -> None:

@ -135,10 +135,10 @@ def _split_share_into_pages(
share = list(enumerate(share_words)) # we need to keep track of the word indices
first = share[:2] # two words on the first page
length = len(share_words)
if length == 12 or length == 20 or length == 24:
if length in (12, 20, 24):
middle = share[2:-2]
last = share[-2:] # two words on the last page
elif length == 33 or length == 18:
elif length in (18, 33):
middle = share[2:]
last = [] # no words at the last page, because it does not add up
else:

@ -48,7 +48,7 @@ def unimport_end(mods: Set[str], collect: bool = True) -> None:
# reallocated at run-time
assert len(sys.modules) <= 160, "Please bump preallocated size in mpconfigport.h"
for mod in sys.modules:
for mod in sys.modules: # pylint: disable=consider-using-dict-items
if mod not in mods:
# remove reference from sys.modules
del sys.modules[mod]
@ -148,7 +148,9 @@ def chunks_intersperse(
if False:
class HashContext(Protocol):
def __init__(self, data: bytes = None) -> None:
def __init__( # pylint: disable=super-init-not-called
self, data: bytes = None
) -> None:
...
def update(self, buf: bytes) -> None:

@ -309,7 +309,7 @@ async def _handle_single_message(
res_msg: protobuf.MessageType | None = None
# We need to find a handler for this message type. Should not raise.
handler = find_handler(ctx.iface, msg.type)
handler = find_handler(ctx.iface, msg.type) # pylint: disable=assignment-from-none
if handler is None:
# If no handler is found, we can skip decoding and directly
@ -431,7 +431,7 @@ async def handle_session(
# Shut down the loop if there is no next message waiting.
# Let the session be restarted from `main`.
loop.clear()
return
return # pylint: disable=lost-exception
except Exception as exc:
# Log and try again. The session handler can only exit explicitly via

@ -24,7 +24,7 @@ INVALID_TYPE = const(-1)
# use it at the same time, thus we check this at runtime in debug builds.
if __debug__:
class BufferLock(object):
class BufferLock:
def __init__(self) -> None:
self.in_use = False
@ -32,17 +32,17 @@ if __debug__:
assert not self.in_use, "global buffer already used by another context"
self.in_use = True
def __exit__(self, type: Any, value: Any, traceback: Any) -> None:
def __exit__(self, exc_type: Any, value: Any, traceback: Any) -> None:
self.in_use = False
else:
class BufferLock(object): # type: ignore
class BufferLock: # type: ignore
def __enter__(self) -> None:
pass
def __exit__(self, type: Any, value: Any, traceback: Any) -> None:
def __exit__(self, exc_type: Any, value: Any, traceback: Any) -> None:
pass

@ -222,5 +222,5 @@ class IdleTimer:
loop.close(task)
"""Global idle timer."""
idle_timer = IdleTimer()
"""Global idle timer."""

@ -137,7 +137,7 @@ def recovery_enter_shares(debug, shares, groups=False):
yield
debug.press_yes()
# Enter shares
for index, share in enumerate(shares):
for share in shares:
br = yield
assert br.code == B.MnemonicInput
# Enter mnemonic words

@ -47,9 +47,9 @@ def _raw_client(request):
return TrezorClientDebugLink(device, auto_interact=not interact)
except Exception:
pass
else:
request.session.shouldstop = "Failed to communicate with Trezor"
raise RuntimeError("No debuggable device found")
request.session.shouldstop = "Failed to communicate with Trezor"
raise RuntimeError("No debuggable device found")
@pytest.fixture(scope="function")

@ -386,7 +386,7 @@ def test_wrong_coordinator(client):
)
with pytest.raises(TrezorFailure, match="Unauthorized operation"):
ownership_proof, _ = btc.get_ownership_proof(
btc.get_ownership_proof(
client,
"Testnet",
parse_path("84'/1'/0'/1/0"),
@ -412,7 +412,7 @@ def test_cancel_authorization(client):
device.cancel_authorization(client)
with pytest.raises(TrezorFailure, match="No preauthorized operation"):
ownership_proof, _ = btc.get_ownership_proof(
btc.get_ownership_proof(
client,
"Testnet",
parse_path("84'/1'/0'/1/0"),

@ -24,11 +24,11 @@ from trezorlib.tools import parse_path
from .. import bip32
def getmultisig(chain, nr, xpubs, signatures=[b"", b"", b""]):
def getmultisig(chain, nr, xpubs):
return messages.MultisigRedeemScriptType(
nodes=[bip32.deserialize(xpub) for xpub in xpubs],
address_n=[chain, nr],
signatures=signatures,
signatures=[b"", b"", b""],
m=2,
)

@ -52,8 +52,8 @@ class TestMsgResetDeviceT2:
yield from click_through(client.debug, screens=18, code=B.ResetDevice)
# show & confirm shares for all groups
for g in range(5):
for h in range(5):
for _g in range(5):
for _h in range(5):
# mnemonic phrases
mnemonic = yield from read_and_confirm_mnemonic(client.debug)
all_mnemonics.append(mnemonic)

@ -48,7 +48,7 @@ def reset_device(client, strength):
yield from click_through(client.debug, screens=8, code=B.ResetDevice)
# show & confirm shares
for h in range(5):
for _ in range(5):
# mnemonic phrases
mnemonic = yield from read_and_confirm_mnemonic(client.debug)
all_mnemonics.append(mnemonic)
@ -141,7 +141,5 @@ def validate_mnemonics(mnemonics, threshold, expected_ems):
assert expected_ems == ems.ciphertext
# We expect these combinations to raise MnemonicError
for test_group in combinations(mnemonics, threshold - 1):
with pytest.raises(
MnemonicError, match=r".*Expected {} mnemonics.*".format(threshold)
):
with pytest.raises(MnemonicError, match=f".*Expected {threshold} mnemonics.*"):
shamir.combine_mnemonics(test_group)

@ -1201,7 +1201,7 @@ class TestMsgSigntx:
setattr(prev_tx, field, value)
name = field.replace("_", " ")
with pytest.raises(
TrezorFailure, match=r"(?i){} not enabled on this coin".format(name)
TrezorFailure, match=rf"(?i){name} not enabled on this coin"
):
btc.sign_tx(
client, "Bitcoin", [inp0], [out1], prev_txes={TXHASH_157041: prev_tx}
@ -1227,7 +1227,7 @@ class TestMsgSigntx:
kwargs = {field: value}
name = field.replace("_", " ")
with pytest.raises(
TrezorFailure, match=r"(?i){} not enabled on this coin".format(name)
TrezorFailure, match=rf"(?i){name} not enabled on this coin"
):
btc.sign_tx(
client, "Bitcoin", [inp0], [out1], prev_txes=TX_CACHE_MAINNET, **kwargs

@ -259,7 +259,7 @@ class TestMsgSigntxBch:
for i in range(1, 4)
]
def getmultisig(chain, nr, signatures=[b"", b"", b""], nodes=nodes):
def getmultisig(chain, nr, signatures):
return proto.MultisigRedeemScriptType(
nodes=nodes, address_n=[chain, nr], signatures=signatures, m=2
)
@ -339,14 +339,16 @@ class TestMsgSigntxBch:
for i in range(1, 4)
]
def getmultisig(chain, nr, signatures=[b"", b"", b""], nodes=nodes):
EMPTY_SIGNATURES = [b"", b"", b""]
def getmultisig(chain, nr, signatures):
return proto.MultisigRedeemScriptType(
nodes=nodes, address_n=[chain, nr], signatures=signatures, m=2
)
inp1 = proto.TxInputType(
address_n=parse_path("48'/145'/3'/0'/0/0"),
multisig=getmultisig(0, 0),
multisig=getmultisig(0, 0, EMPTY_SIGNATURES),
amount=48490,
prev_hash=TXHASH_8b6db9,
prev_index=0,
@ -359,7 +361,7 @@ class TestMsgSigntxBch:
)
out2 = proto.TxOutputType(
address_n=parse_path("48'/145'/3'/0'/1/0"),
multisig=getmultisig(1, 0),
multisig=getmultisig(1, 0, EMPTY_SIGNATURES),
script_type=proto.OutputScriptType.PAYTOMULTISIG,
amount=24000,
)

@ -202,14 +202,16 @@ class TestMsgSigntxBitcoinGold:
for i in range(1, 4)
]
def getmultisig(chain, nr, signatures=[b"", b"", b""], nodes=nodes):
EMPTY_SIGS = [b"", b"", b""]
def getmultisig(chain, nr, signatures):
return proto.MultisigRedeemScriptType(
nodes=nodes, address_n=[chain, nr], signatures=signatures, m=2
)
inp1 = proto.TxInputType(
address_n=parse_path("48'/156'/3'/0'/0/0"),
multisig=getmultisig(0, 0),
multisig=getmultisig(0, 0, EMPTY_SIGS),
# 33Ju286QvonBz5N1V754ZekQv4GLJqcc5R
amount=1252382934,
prev_hash=TXHASH_25526b,
@ -223,7 +225,7 @@ class TestMsgSigntxBitcoinGold:
)
out2 = proto.TxOutputType(
address_n=parse_path("48'/156'/3'/0'/1/0"),
multisig=getmultisig(1, 0),
multisig=getmultisig(1, 0, EMPTY_SIGS),
script_type=proto.OutputScriptType.PAYTOMULTISIG,
amount=1252382934 - 24000 - 1000,
)

@ -982,7 +982,7 @@ def test_p2wpkh_invalid_signature(client):
}
with pytest.raises(TrezorFailure, match="Invalid signature"):
_, serialized_tx = btc.sign_tx(
btc.sign_tx(
client,
"Testnet",
[inp1],

@ -224,8 +224,6 @@ def test_signtx_multisig(client, paths, address_index):
multisig=multisig,
)
sig, serialized_tx = btc.sign_tx(
client, "Bitcoin", [inp1], [out1], prev_txes=TX_CACHE_MAINNET
)
sig, _ = btc.sign_tx(client, "Bitcoin", [inp1], [out1], prev_txes=TX_CACHE_MAINNET)
assert sig[0]

@ -129,6 +129,4 @@ class TestOpReturn:
with pytest.raises(
TrezorFailure, match="Output's address_n provided but not expected."
):
_, serialized_tx = btc.sign_tx(
client, "Bitcoin", [inp1], [out1], prev_txes=TX_API
)
btc.sign_tx(client, "Bitcoin", [inp1], [out1], prev_txes=TX_API)

@ -75,8 +75,8 @@ def reset(client, strength=128):
yield from click_through(client.debug, screens=18, code=B.ResetDevice)
# show & confirm shares for all groups
for g in range(5):
for h in range(5):
for _g in range(5):
for _h in range(5):
# mnemonic phrases
mnemonic = yield from read_and_confirm_mnemonic(client.debug)
all_mnemonics.append(mnemonic)

@ -59,7 +59,7 @@ def reset(client, strength=128):
yield from click_through(client.debug, screens=8, code=B.ResetDevice)
# show & confirm shares
for h in range(5):
for _ in range(5):
# mnemonic phrases
mnemonic = yield from read_and_confirm_mnemonic(client.debug)
all_mnemonics.append(mnemonic)

Loading…
Cancel
Save