What changed, and why it matters
This is a code cleanup and type-safety refactor for the Monero parts of the Trezor firmware. It adds stricter type hints, asserts that certain values are not missing, and fixes a minor return-value bug in a helper that encrypts data. There is no direct evidence this commit fixes an active security vulnerability, but it reduces the chance of future bugs by making assumptions explicit.
Treat as a normal maintenance refactor. Reviewers may optionally verify that the new `assert` statements do not introduce denial-of-service paths from malformed host messages, but the changes appear defensive and consistent with Trezor's existing validation style.
Security signals we found
assert guards added to prevent use of None values in Monero address and transaction handling
chacha_poly.encrypt return signature corrected from 3-tuple to 2-tuple, with caller updated
type annotations tightened across Monero signing, key image sync, and bulletproof code
bytes/bytearray/memoryview inputs accepted more consistently via AnyBytes/AnyBuffer aliases
Evidence from the diff
The commit refactors Monero-related Python code in the Trezor Core firmware. Changes include: replacing generic bytes annotations with AnyBytes/AnyBuffer aliases; adding assert guards for fields that must be non-None (e.g., creds.address, dst.addr, tsx_data.fee, tsx_data.num_inputs, state.tx_prefix_hasher); changing chacha_poly.encrypt to return a 2-tuple (nonce, ciphertext+tag) instead of a 3-tuple and updating its caller; fixing a few type-only imports and import paths; and making some function signatures stricter (e.g., real_output_index: int instead of int | None). The commit is titled as a refactor and explicitly says ‘[no changelog]’.
Changed components
core/src/apps/monero/get_address.pycore/src/apps/monero/get_watch_only.pycore/src/apps/monero/key_image_sync.pycore/src/apps/monero/layout.pycore/src/apps/monero/live_refresh.pycore/src/apps/monero/misc.pycore/src/apps/monero/signing/offloading_keys.pycore/src/apps/monero/signing/state.pycore/src/apps/monero/signing/step_01_init_transaction.pycore/src/apps/monero/xmr/addresses.pycore/src/apps/monero/xmr/bulletproof.pycore/src/apps/monero/xmr/chacha_poly.pycore/src/apps/monero/xmr/crypto_helpers.pycore/src/apps/monero/xmr/keccak_hasher.pycore/src/apps/monero/xmr/key_image.pycore/src/apps/monero/xmr/mlsag_hasher.pycore/src/apps/monero/xmr/monero.pycore/src/apps/monero/xmr/networks.pyInspect captured patch +180 / −107
diff --git a/core/src/apps/monero/get_address.py b/core/src/apps/monero/get_address.py
index 99b3e83b..9ddd3969 100644
--- a/core/src/apps/monero/get_address.py
+++ b/core/src/apps/monero/get_address.py
@@ -27,7 +27,8 @@ async def get_address(msg: MoneroGetAddress, keychain: Keychain) -> MoneroAddres
await paths.validate_path(keychain, address_n)
creds = misc.get_creds(keychain, address_n, msg.network_type)
- addr = creds.address
+ address = creds.address
+ assert address is not None
have_subaddress = (
account is not None and minor is not None and (account, minor) != (0, 0)
@@ -44,7 +45,7 @@ async def get_address(msg: MoneroGetAddress, keychain: Keychain) -> MoneroAddres
assert payment_id is not None
if len(payment_id) != 8:
raise ValueError("Invalid payment ID length")
- addr = addresses.encode_addr(
+ address = addresses.encode_addr(
net_version(msg.network_type, False, True),
crypto_helpers.encodepoint(creds.spend_key_public),
crypto_helpers.encodepoint(creds.view_key_public),
@@ -59,7 +60,7 @@ async def get_address(msg: MoneroGetAddress, keychain: Keychain) -> MoneroAddres
creds.view_key_private, creds.spend_key_public, account, minor
)
- addr = addresses.encode_addr(
+ address = addresses.encode_addr(
net_version(msg.network_type, True, False),
crypto_helpers.encodepoint(pub_spend),
crypto_helpers.encodepoint(pub_view),
@@ -70,12 +71,12 @@ async def get_address(msg: MoneroGetAddress, keychain: Keychain) -> MoneroAddres
coin = "XMR"
await show_address(
- addr,
+ address,
subtitle=TR.address__coin_address_template.format(coin),
- address_qr="monero:" + addr,
+ address_qr="monero:" + address,
path=paths.address_n_to_str(address_n),
account=paths.get_account_name(coin, msg.address_n, PATTERN, SLIP44_ID),
chunkify=bool(msg.chunkify),
)
- return MoneroAddress(address=addr.encode())
+ return MoneroAddress(address=address.encode())
diff --git a/core/src/apps/monero/get_watch_only.py b/core/src/apps/monero/get_watch_only.py
index 322f05d0..49834160 100644
--- a/core/src/apps/monero/get_watch_only.py
+++ b/core/src/apps/monero/get_watch_only.py
@@ -22,6 +22,7 @@ async def get_watch_only(msg: MoneroGetWatchKey, keychain: Keychain) -> MoneroWa
creds = misc.get_creds(keychain, msg.address_n, msg.network_type)
address = creds.address
+ assert address is not None
watch_key = crypto_helpers.encodeint(creds.view_key_private)
return MoneroWatchKey(watch_key=watch_key, address=address.encode())
diff --git a/core/src/apps/monero/key_image_sync.py b/core/src/apps/monero/key_image_sync.py
index b107be98..3e663fc8 100644
--- a/core/src/apps/monero/key_image_sync.py
+++ b/core/src/apps/monero/key_image_sync.py
@@ -6,6 +6,9 @@ from apps.common.keychain import auto_keychain
from apps.monero import layout
if TYPE_CHECKING:
+ from buffer_types import AnyBytes
+
+ from trezor.crypto.hashlib import sha3_256
from trezor.messages import (
MoneroKeyImageExportInitAck,
MoneroKeyImageExportInitRequest,
@@ -13,7 +16,7 @@ if TYPE_CHECKING:
MoneroKeyImageSyncStepAck,
MoneroKeyImageSyncStepRequest,
)
- from trezor.ui.layouts.common import ProgressLayout
+ from trezor.ui import ProgressLayout
from apps.common.keychain import Keychain
@@ -56,13 +59,13 @@ class KeyImageSync:
def __init__(self):
from apps.monero.xmr import crypto_helpers
- self.current_output = -1
- self.num_outputs = 0
- self.expected_hash = b""
- self.enc_key = b""
+ self.current_output: int = -1
+ self.num_outputs: int = 0
+ self.expected_hash: AnyBytes = b""
+ self.enc_key: AnyBytes = b""
self.creds: AccountCreds | None = None
self.subaddresses = {}
- self.hasher = crypto_helpers.get_keccak()
+ self.hasher: sha3_256 = crypto_helpers.get_keccak()
async def _init_step(
@@ -137,7 +140,7 @@ def _sync_step(
crypto.encodeint_into(buff_mv[64:], sig[0][1])
# Encrypt with enc_key
- nonce, ciph, _ = chacha_poly.encrypt(s.enc_key, buff)
+ nonce, ciph = chacha_poly.encrypt(s.enc_key, buff)
kis.append(MoneroExportedKeyImage(iv=nonce, blob=ciph))
diff --git a/core/src/apps/monero/layout.py b/core/src/apps/monero/layout.py
index 8c79c2e7..48703315 100644
--- a/core/src/apps/monero/layout.py
+++ b/core/src/apps/monero/layout.py
@@ -13,6 +13,8 @@ DUMMY_PAYMENT_ID = b"\x00\x00\x00\x00\x00\x00\x00\x00"
if TYPE_CHECKING:
+ from buffer_types import AnyBytes
+
from trezor.enums import MoneroNetworkType
from trezor.messages import MoneroTransactionData, MoneroTransactionDestinationEntry
@@ -117,7 +119,7 @@ async def require_confirm_transaction(
change_idx = get_change_addr_idx(outputs, tsx_data.change_dts)
payment_id = tsx_data.payment_id # local_cache_attribute
- if tsx_data.unlock_time != 0:
+ if tsx_data.unlock_time is not None and tsx_data.unlock_time != 0:
await _require_confirm_unlock_time(tsx_data.unlock_time)
for idx, dst in enumerate(outputs):
@@ -142,6 +144,7 @@ async def require_confirm_transaction(
):
await _require_confirm_payment_id(payment_id)
+ assert tsx_data.fee is not None
await _require_confirm_fee(tsx_data.fee)
progress.step(state, 0)
@@ -149,7 +152,7 @@ async def require_confirm_transaction(
async def _require_confirm_output(
dst: MoneroTransactionDestinationEntry,
network_type: MoneroNetworkType,
- payment_id: bytes | None,
+ payment_id: AnyBytes | None,
chunkify: bool,
) -> None:
"""
@@ -160,7 +163,14 @@ async def _require_confirm_output(
from apps.monero.xmr.addresses import encode_addr
from apps.monero.xmr.networks import net_version
- version = net_version(network_type, dst.is_subaddress, payment_id is not None)
+ assert dst.addr is not None
+ assert dst.addr.spend_public_key is not None
+ assert dst.addr.view_public_key is not None
+ assert dst.amount is not None
+
+ version = net_version(
+ network_type, dst.is_subaddress or False, payment_id is not None
+ )
addr = encode_addr(
version, dst.addr.spend_public_key, dst.addr.view_public_key, payment_id
)
@@ -173,7 +183,7 @@ async def _require_confirm_output(
)
-async def _require_confirm_payment_id(payment_id: bytes) -> None:
+async def _require_confirm_payment_id(payment_id: AnyBytes) -> None:
from trezor.ui.layouts import confirm_blob
await confirm_blob(
diff --git a/core/src/apps/monero/live_refresh.py b/core/src/apps/monero/live_refresh.py
index eb43ad4e..e30a5f80 100644
--- a/core/src/apps/monero/live_refresh.py
+++ b/core/src/apps/monero/live_refresh.py
@@ -11,7 +11,7 @@ if TYPE_CHECKING:
MoneroLiveRefreshStepAck,
MoneroLiveRefreshStepRequest,
)
- from trezor.ui.layouts.common import ProgressLayout
+ from trezor.ui import ProgressLayout
from apps.common.keychain import Keychain
diff --git a/core/src/apps/monero/misc.py b/core/src/apps/monero/misc.py
index d6af6cbc..b6122a16 100644
--- a/core/src/apps/monero/misc.py
+++ b/core/src/apps/monero/misc.py
@@ -1,6 +1,8 @@
from typing import TYPE_CHECKING
if TYPE_CHECKING:
+ from buffer_types import AnyBytes
+
from trezor.enums import MoneroNetworkType
from apps.common.keychain import Keychain
@@ -27,12 +29,13 @@ def get_creds(
def compute_tx_key(
spend_key_private: Scalar,
- tx_prefix_hash: bytes,
- salt: bytes,
+ tx_prefix_hash: AnyBytes,
+ salt: AnyBytes,
rand_mult_num: Scalar,
) -> bytes:
from apps.monero.xmr import crypto, crypto_helpers
+ tx_prefix_hash = bytes(tx_prefix_hash)
rand_inp = crypto.sc_add_into(None, spend_key_private, rand_mult_num)
passwd = crypto_helpers.keccak_2hash(
crypto_helpers.encodeint(rand_inp) + tx_prefix_hash
@@ -42,13 +45,14 @@ def compute_tx_key(
def compute_enc_key_host(
- view_key_private: Scalar, tx_prefix_hash: bytes
+ view_key_private: Scalar, tx_prefix_hash: AnyBytes
) -> tuple[bytes, bytes]:
from trezor.crypto import random
from apps.monero.xmr import crypto_helpers
salt = random.bytes(32)
+ tx_prefix_hash = bytes(tx_prefix_hash)
passwd = crypto_helpers.keccak_2hash(
crypto_helpers.encodeint(view_key_private) + tx_prefix_hash
)
diff --git a/core/src/apps/monero/signing/offloading_keys.py b/core/src/apps/monero/signing/offloading_keys.py
index 252adf39..60cabdfe 100644
--- a/core/src/apps/monero/signing/offloading_keys.py
+++ b/core/src/apps/monero/signing/offloading_keys.py
@@ -4,6 +4,8 @@ from typing import TYPE_CHECKING
from apps.monero.xmr.crypto_helpers import compute_hmac
if TYPE_CHECKING:
+ from buffer_types import AnyBuffer, AnyBytes
+
from trezor.messages import (
MoneroTransactionDestinationEntry,
MoneroTransactionSourceEntry,
@@ -17,11 +19,11 @@ _BUILD_KEY_BUFFER = bytearray(_SECRET_LENGTH + _DISCRIMINATOR_LENGTH + _INDEX_LE
def _build_key(
- secret: bytes,
- discriminator: bytes,
+ secret: AnyBytes,
+ discriminator: AnyBytes,
index: int | None = None,
- out: bytes | None = None,
-) -> bytes:
+ out: AnyBuffer | None = None,
+) -> AnyBuffer:
"""
Creates an unique-purpose key
"""
@@ -54,49 +56,49 @@ def _build_key(
return crypto_helpers.keccak_2hash(key_buff, out)
-def hmac_key_txin(key_hmac: bytes, idx: int) -> bytes:
+def hmac_key_txin(key_hmac: AnyBytes, idx: int) -> AnyBuffer:
"""
(TxSourceEntry[i] || tx.vin[i]) hmac key
"""
return _build_key(key_hmac, b"txin", idx)
-def hmac_key_txin_comm(key_hmac: bytes, idx: int) -> bytes:
+def hmac_key_txin_comm(key_hmac: AnyBytes, idx: int) -> AnyBuffer:
"""
pseudo_outputs[i] hmac key. Pedersen commitment for inputs.
"""
return _build_key(key_hmac, b"txin-comm", idx)
-def _hmac_key_txdst(key_hmac: bytes, idx: int) -> bytes:
+def _hmac_key_txdst(key_hmac: AnyBytes, idx: int) -> AnyBuffer:
"""
TxDestinationEntry[i] hmac key
"""
return _build_key(key_hmac, b"txdest", idx)
-def _hmac_key_txout(key_hmac: bytes, idx: int) -> bytes:
+def _hmac_key_txout(key_hmac: AnyBytes, idx: int) -> AnyBuffer:
"""
(TxDestinationEntry[i] || tx.vout[i]) hmac key
"""
return _build_key(key_hmac, b"txout", idx)
-def enc_key_txin_alpha(key_enc: bytes, idx: int) -> bytes:
+def enc_key_txin_alpha(key_enc: AnyBytes, idx: int) -> AnyBuffer:
"""
Chacha20Poly1305 encryption key for alpha[i] used in Pedersen commitment in pseudo_outs[i]
"""
return _build_key(key_enc, b"txin-alpha", idx)
-def enc_key_spend(key_enc: bytes, idx: int) -> bytes:
+def enc_key_spend(key_enc: AnyBytes, idx: int) -> AnyBuffer:
"""
Chacha20Poly1305 encryption key for alpha[i] used in Pedersen commitment in pseudo_outs[i]
"""
return _build_key(key_enc, b"txin-spend", idx)
-def key_signature(master: bytes, idx: int, is_iv: bool = False) -> bytes:
+def key_signature(master: AnyBytes, idx: int, is_iv: bool = False) -> AnyBuffer:
"""
Generates signature offloading related offloading keys
"""
@@ -104,7 +106,7 @@ def key_signature(master: bytes, idx: int, is_iv: bool = False) -> bytes:
def gen_hmac_vini(
- key: bytes, src_entr: MoneroTransactionSourceEntry, vini_bin: bytes, idx: int
+ key: AnyBytes, src_entr: MoneroTransactionSourceEntry, vini_bin: AnyBytes, idx: int
) -> bytes:
"""
Computes hmac (TxSourceEntry[i] || tx.vin[i])
@@ -122,6 +124,10 @@ def gen_hmac_vini(
kwriter = get_keccak_writer()
real_outputs = src_entr.outputs
real_additional = src_entr.real_out_additional_tx_keys
+
+ assert src_entr.real_output is not None
+ assert src_entr.real_output_in_tx_index is not None
+
src_entr.outputs = [src_entr.outputs[src_entr.real_output]]
if real_additional and len(real_additional) > 1:
src_entr.real_out_additional_tx_keys = [
@@ -139,7 +145,10 @@ def gen_hmac_vini(
def gen_hmac_vouti(
- key: bytes, dst_entr: MoneroTransactionDestinationEntry, tx_out_bin: bytes, idx: int
+ key: AnyBytes,
+ dst_entr: MoneroTransactionDestinationEntry,
+ tx_out_bin: AnyBytes,
+ idx: int,
) -> bytes:
"""
Generates HMAC for (TxDestinationEntry[i] || tx.vout[i])
@@ -158,7 +167,7 @@ def gen_hmac_vouti(
def gen_hmac_tsxdest(
- key: bytes, dst_entr: MoneroTransactionDestinationEntry, idx: int
+ key: AnyBytes, dst_entr: MoneroTransactionDestinationEntry, idx: int
) -> bytes:
"""
Generates HMAC for TxDestinationEntry[i]
@@ -175,7 +184,7 @@ def gen_hmac_tsxdest(
return hmac_tsxdest
-def get_ki_from_vini(vini_bin: bytes) -> bytes:
+def get_ki_from_vini(vini_bin: AnyBytes) -> bytes:
"""
Returns key image from the TxinToKey, which is currently
serialized as the last 32 bytes.
diff --git a/core/src/apps/monero/signing/state.py b/core/src/apps/monero/signing/state.py
index ad9e3bde..c9bd2bc8 100644
--- a/core/src/apps/monero/signing/state.py
+++ b/core/src/apps/monero/signing/state.py
@@ -1,12 +1,13 @@
from typing import TYPE_CHECKING
if TYPE_CHECKING:
+ from buffer_types import AnyBytes
+
from trezor.messages import MoneroTransactionDestinationEntry
from apps.monero.xmr.credentials import AccountCreds
- from apps.monero.xmr.crypto import Point, Scalar
- Subaddresses = dict[bytes, tuple[int, int]]
+ Subaddresses = dict[AnyBytes, tuple[int, int]]
class State:
@@ -32,8 +33,8 @@ class State:
self.creds: AccountCreds | None = None
# HMAC/encryption keys used to protect offloaded data
- self.key_hmac: bytes | None = None
- self.key_enc: bytes | None = None
+ self.key_hmac: AnyBytes
+ self.key_enc: AnyBytes | None = None
"""
Transaction keys
@@ -43,8 +44,8 @@ class State:
- for subaddresses the `r` is commonly denoted as `s`, however it is still just a random number
- the keys are used to derive the one time address and its keys (P = H(A*r)*G + B)
"""
- self.tx_priv: Scalar | None = None
- self.tx_pub: Point | None = None
+ self.tx_priv: crypto.Scalar
+ self.tx_pub: crypto.Point | None = None
"""
In some cases when subaddresses are used we need more tx_keys
@@ -56,21 +57,21 @@ class State:
self.client_version = 0
self.hard_fork = 12
- self.input_count: int | None = 0
+ self.input_count: int = 0
self.output_count = 0
self.progress_total = 0
self.progress_cur = 0
- self.output_change: "MoneroTransactionDestinationEntry" | None = None
- self.fee: int | None = 0
+ self.output_change: MoneroTransactionDestinationEntry | None = None
+ self.fee: int = 0
self.tx_type = 0
# wallet sub-address major index
self.account_idx: int | None = 0
# contains additional tx keys if need_additional_tx_keys is True
- self.additional_tx_private_keys: list[Scalar] = []
- self.additional_tx_public_keys: list[bytes] | None = []
+ self.additional_tx_private_keys: list[crypto.Scalar] = []
+ self.additional_tx_public_keys: list[AnyBytes] | None = []
# currently processed input/output index
self.current_input_index = -1
@@ -84,11 +85,11 @@ class State:
self.summary_outs_money: int | None = 0
# output commitments
- self.output_pk_commitments: list[bytes] | None = []
+ self.output_pk_commitments: list[AnyBytes] | None = []
self.output_amounts: list[int] | None = []
# output *range proof* masks. HP10+ makes them deterministic.
- self.output_masks: list[Scalar] | None = []
+ self.output_masks: list[crypto.Scalar] | None = []
# the range proofs are calculated in batches, this denotes the grouping
self.rsig_grouping: list[int] | None = []
@@ -96,9 +97,9 @@ class State:
self.rsig_offload: bool | None = False
# sum of all inputs' pseudo out masks
- self.sumpouts_alphas: Scalar = crypto.Scalar(0)
+ self.sumpouts_alphas: crypto.Scalar = crypto.Scalar(0)
# sum of all output' pseudo out masks
- self.sumout: Scalar = crypto.Scalar(0)
+ self.sumout: crypto.Scalar = crypto.Scalar(0)
self.subaddresses: Subaddresses | None = {}
diff --git a/core/src/apps/monero/signing/step_01_init_transaction.py b/core/src/apps/monero/signing/step_01_init_transaction.py
index b4fa7422..6eda38fe 100644
--- a/core/src/apps/monero/signing/step_01_init_transaction.py
+++ b/core/src/apps/monero/signing/step_01_init_transaction.py
@@ -10,6 +10,7 @@ from apps.monero.xmr import crypto, crypto_helpers, monero
if TYPE_CHECKING:
from trezor.messages import (
MoneroAccountPublicAddress,
+ MoneroNetworkType,
MoneroTransactionData,
MoneroTransactionDestinationEntry,
MoneroTransactionInitAck,
@@ -23,7 +24,7 @@ if TYPE_CHECKING:
async def init_transaction(
state: State,
address_n: list,
- network_type: int,
+ network_type: MoneroNetworkType,
tsx_data: MoneroTransactionData,
keychain,
progress: MoneroTransactionProgress,
@@ -46,16 +47,18 @@ async def init_transaction(
state.fee = state.fee if state.fee > 0 else 0
state.tx_priv = crypto.random_scalar()
+ assert state.tx_priv is not None
state.tx_pub = crypto.scalarmult_base_into(None, state.tx_priv)
mem_trace(1)
+ assert tsx_data.num_inputs is not None
state.input_count = tsx_data.num_inputs
state.output_count = len(outputs)
- assert state.input_count is not None
state.progress_total = 4 + 3 * state.input_count + state.output_count
state.progress_cur = 0
# Ask for confirmation
+ assert state.creds.network_type is not None
await layout.require_confirm_transaction(
state,
tsx_data,
@@ -69,6 +72,7 @@ async def init_transaction(
# Basic transaction parameters
state.output_change = tsx_data.change_dts
+ assert tsx_data.fee is not None
state.fee = tsx_data.fee
state.account_idx = tsx_data.account
state.last_step = state.STEP_INIT
@@ -85,6 +89,7 @@ async def init_transaction(
if state.output_count < 2:
raise signing.NotEnoughOutputsError("At least two outputs are required")
+ assert tsx_data.rsig_data is not None
_check_rsig_data(state, tsx_data.rsig_data)
_check_subaddresses(state, outputs)
@@ -94,6 +99,8 @@ async def init_transaction(
gc.collect()
# Iterative tx_prefix_hash hash computation
+ assert state.tx_prefix_hasher is not None
+ assert tsx_data.unlock_time is not None
state.tx_prefix_hasher.uvarint(2) # current Monero transaction format (RingCT = 2)
state.tx_prefix_hasher.uvarint(tsx_data.unlock_time)
state.tx_prefix_hasher.uvarint(state.input_count) # ContainerType, size
diff --git a/core/src/apps/monero/xmr/addresses.py b/core/src/apps/monero/xmr/addresses.py
index d7679b83..fc167540 100644
--- a/core/src/apps/monero/xmr/addresses.py
+++ b/core/src/apps/monero/xmr/addresses.py
@@ -1,6 +1,8 @@
from typing import TYPE_CHECKING
if TYPE_CHECKING:
+ from buffer_types import AnyBytes
+
from trezor.messages import (
MoneroAccountPublicAddress,
MoneroTransactionDestinationEntry,
@@ -8,14 +10,14 @@ if TYPE_CHECKING:
def encode_addr(
- version, spend_pub: bytes, view_pub: bytes, payment_id: bytes | None = None
+ version, spend_pub: AnyBytes, view_pub: AnyBytes, payment_id: AnyBytes | None = None
) -> str:
"""
Builds Monero address from public keys
"""
from trezor.crypto import monero as tcry
- buf = spend_pub + view_pub
+ buf = bytes(spend_pub) + bytes(view_pub)
if payment_id:
buf += bytes(payment_id)
return tcry.xmr_base58_addr_encode_check(ord(version), bytes(buf))
@@ -34,11 +36,14 @@ def classify_subaddresses(
addr_set = set()
for tx in tx_dests:
addr = tx.addr # local_cache_attribute
+ assert addr is not None
if change_addr and addr_eq(change_addr, addr):
continue
+ assert addr.spend_public_key is not None
+ assert addr.view_public_key is not None
# addr_to_hash
# Creates hashable address representation
- addr_hashed = bytes(addr.spend_public_key + addr.view_public_key)
+ addr_hashed = bytes(addr.spend_public_key) + bytes(addr.view_public_key)
if addr_hashed in addr_set:
continue
addr_set.add(addr_hashed)
@@ -67,8 +72,10 @@ def get_change_addr_idx(
if change_dts is None:
return None
+ assert change_dts.addr is not None
change_idx = None
for idx, dst in enumerate(outputs):
+ assert dst.addr is not None
if (
change_dts.amount
and change_dts.amount == dst.amount
diff --git a/core/src/apps/monero/xmr/bulletproof.py b/core/src/apps/monero/xmr/bulletproof.py
index 6d5d4a51..455cc12f 100644
--- a/core/src/apps/monero/xmr/bulletproof.py
+++ b/core/src/apps/monero/xmr/bulletproof.py
@@ -8,6 +8,7 @@ from apps.monero.xmr import crypto, crypto_helpers
from apps.monero.xmr.serialize.int_serialize import dump_uvarint_b_into
if TYPE_CHECKING:
+ from buffer_types import AnyBuffer, AnyBytes
from typing import Generic, Iterator, TypeVar
from .serialize_messages.tx_rsig_bulletproof import BulletproofPlus
@@ -86,12 +87,15 @@ def _ensure_dst_key(dst: bytearray | None = None) -> bytearray:
def memcpy(
- dst: bytearray, dst_off: int, src: bytes, src_off: int, len: int
+ dst: bytearray,
+ dst_off: int,
+ src: AnyBytes,
+ src_off: int,
+ length: int,
) -> bytearray:
from trezor.utils import memcpy as tmemcpy
- if dst is not None:
- tmemcpy(dst, dst_off, src, src_off, len)
+ tmemcpy(dst, dst_off, src, src_off, length)
return dst
@@ -111,7 +115,7 @@ def _gc_iter(i: int) -> None:
gc_collect()
-def _invert(dst: bytearray | None, x: bytes) -> bytearray:
+def _invert(dst: bytearray | None, x: AnyBytes) -> bytearray:
dst = _ensure_dst_key(dst)
decodeint_into_noreduce(_tmp_sc_1, x)
sc_inv_into(_tmp_sc_2, _tmp_sc_1)
@@ -163,7 +167,7 @@ def _sc_gen(dst: bytearray | None = None) -> bytearray:
return dst
-def _sc_add(dst: bytearray | None, a: bytes, b: bytes) -> bytearray:
+def _sc_add(dst: bytearray | None, a: AnyBytes, b: AnyBytes) -> bytearray:
dst = _ensure_dst_key(dst)
decodeint_into_noreduce(_tmp_sc_1, a)
decodeint_into_noreduce(_tmp_sc_2, b)
@@ -174,8 +178,8 @@ def _sc_add(dst: bytearray | None, a: bytes, b: bytes) -> bytearray:
def _sc_sub(
dst: bytearray | None,
- a: bytes | Scalar,
- b: bytes | Scalar,
+ a: AnyBytes | Scalar,
+ b: AnyBytes | Scalar,
) -> bytearray:
dst = _ensure_dst_key(dst)
@@ -190,7 +194,7 @@ def _sc_sub(
return dst
-def _sc_mul(dst: bytearray | None, a: bytes, b: bytes | Scalar) -> bytearray:
+def _sc_mul(dst: bytearray | None, a: AnyBytes, b: AnyBytes | Scalar) -> bytearray:
dst = _ensure_dst_key(dst)
decodeint_into_noreduce(_tmp_sc_1, a)
@@ -204,9 +208,9 @@ def _sc_mul(dst: bytearray | None, a: bytes, b: bytes | Scalar) -> bytearray:
def _sc_muladd(
dst: ScalarDst | None,
- a: bytes | Scalar,
- b: bytes | Scalar,
- c: bytes | Scalar,
+ a: AnyBytes | Scalar,
+ b: AnyBytes | Scalar,
+ c: AnyBytes | Scalar,
) -> ScalarDst:
if isinstance(dst, Scalar):
dst_sc = dst
@@ -228,7 +232,7 @@ def _sc_muladd(
return dst
-def _add_keys(dst: bytearray | None, A: bytes, B: bytes) -> bytearray:
+def _add_keys(dst: bytearray | None, A: AnyBytes, B: AnyBytes) -> bytearray:
dst = _ensure_dst_key(dst)
decodepoint_into(_tmp_pt_1, A)
decodepoint_into(_tmp_pt_2, B)
@@ -237,7 +241,9 @@ def _add_keys(dst: bytearray | None, A: bytes, B: bytes) -> bytearray:
return dst
-def _add_keys2(dst: bytearray | None, a: bytes, b: bytes, B: bytes) -> bytearray:
+def _add_keys2(
+ dst: bytearray | None, a: AnyBytes, b: AnyBytes, B: AnyBytes
+) -> bytearray:
dst = _ensure_dst_key(dst)
decodeint_into_noreduce(_tmp_sc_1, a)
decodeint_into_noreduce(_tmp_sc_2, b)
@@ -429,11 +435,13 @@ class KeyVBase(Generic[T]):
def __len__(self) -> int:
return self.size
- def to(self, idx: int, buff: bytearray | None = None, offset: int = 0) -> bytearray:
+ def to(
+ self, idx: int, buff: bytearray | None = None, offset: int = 0
+ ) -> bytearray | None:
buff = _ensure_dst_key(buff)
return memcpy(buff, offset, self.__getitem__(self.idxize(idx)), 0, 32)
- def read(self, idx: int, buff: bytes, offset: int = 0) -> bytes:
+ def read(self, idx: int, buff: AnyBytes, offset: int = 0) -> None:
raise NotImplementedError
def slice_view(self, start: int, stop: int) -> "KeyVSliced":
@@ -506,7 +514,7 @@ class KeyV(KeyVBaseType[T]):
assert isinstance(self.d, TBYTES)
self.mv = memoryview(self.d)
- def __getitem__(self, item):
+ def __getitem__(self, item: int) -> AnyBuffer:
"""
Returns corresponding 32 byte array.
Creates new memoryview on access.
@@ -517,7 +525,7 @@ class KeyV(KeyVBaseType[T]):
assert self.mv is not None
return self.mv[item * 32 : (item + 1) * 32]
- def __setitem__(self, key, value):
+ def __setitem__(self, key: int, value: AnyBytes) -> None:
if self.chunked:
self.read(key, value)
if self.const:
@@ -544,7 +552,7 @@ class KeyV(KeyVBaseType[T]):
memcpy(buff if buff else self.cur, offset, d, idx << 5, 32)
return buff if buff else self.cur
- def read(self, idx: int, buff: bytes, offset: int = 0) -> bytes:
+ def read(self, idx: int, buff: AnyBytes, offset: int = 0) -> None:
d = self.d # local_cache_attribute
idx = self.idxize(idx)
@@ -629,7 +637,7 @@ class KeyVEval(KeyVBase):
self.scalar = scalar
self.buff = _ensure_dst_key() if not raw else (Scalar() if scalar else Point())
- def __getitem__(self, item):
+ def __getitem__(self, item: int):
return self.fnc(self.idxize(item), self.buff)
def to(self, idx, buff: bytearray | None = None, offset: int = 0):
@@ -645,7 +653,7 @@ class KeyVEval(KeyVBase):
return self_buff
else:
raise ValueError("Not supported")
- else:
+ elif buff is not None:
memcpy(buff, offset, self_buff, 0, 32)
return buff if buff else self_buff
@@ -678,8 +686,11 @@ class KeyVPrecomp(KeyVBase):
if item < len(self.precomp_prefix):
return self.precomp_prefix.to(item, buff if buff else self_buff, offset)
self.aux_comp_fnc(item, self_buff)
- memcpy(buff, offset, self_buff, 0, 32)
- return buff if buff else self_buff
+ if buff:
+ memcpy(buff, offset, self_buff, 0, 32)
+ return buff
+ else:
+ return self_buff
class KeyVSliced(KeyVBase):
@@ -1054,11 +1065,10 @@ class KeyChallengeCacheVct(KeyVBase):
return cur
-def _ensure_dst_keyvect(dst=None, size: int | None = None):
+def _ensure_dst_keyvect(dst: KeyV | None, size: int):
if dst is None:
- dst = KeyV(elems=size)
- return dst
- if size is not None and size != len(dst):
+ return KeyV(elems=size)
+ elif size != len(dst):
dst.resize(size)
return dst
diff --git a/core/src/apps/monero/xmr/chacha_poly.py b/core/src/apps/monero/xmr/chacha_poly.py
index eb93e333..9d8d65c7 100644
--- a/core/src/apps/monero/xmr/chacha_poly.py
+++ b/core/src/apps/monero/xmr/chacha_poly.py
@@ -1,11 +1,18 @@
+from typing import TYPE_CHECKING
+
from trezor.crypto import (
AuthenticationError,
chacha20poly1305_decrypt,
chacha20poly1305_encrypt,
)
+if TYPE_CHECKING:
+ from buffer_types import AnyBytes
+
-def encrypt(key: bytes, plaintext: bytes, associated_data: bytes | None = None):
+def encrypt(
+ key: AnyBytes, plaintext: AnyBytes, associated_data: AnyBytes | None = None
+) -> tuple[bytes, bytes]:
"""
Uses ChaCha20Poly1305 for encryption
"""
@@ -17,15 +24,15 @@ def encrypt(key: bytes, plaintext: bytes, associated_data: bytes | None = None):
cipher.auth(associated_data)
ciphertext = cipher.encrypt(plaintext)
tag = cipher.finish()
- return nonce, ciphertext + tag, b""
+ return nonce, ciphertext + tag
def _decrypt(
- key: bytes,
- iv: bytes,
- ciphertext: bytes,
- associated_data: bytes | None = None,
-):
+ key: AnyBytes,
+ iv: AnyBytes,
+ ciphertext: AnyBytes,
+ associated_data: AnyBytes | None = None,
+) -> bytes:
"""
ChaCha20Poly1305 decryption
"""
@@ -43,12 +50,14 @@ def _decrypt(
return plaintext
-def encrypt_pack(key: bytes, plaintext: bytes, associated_data: bytes | None = None):
+def encrypt_pack(
+ key: AnyBytes, plaintext: AnyBytes, associated_data: AnyBytes | None = None
+) -> bytes:
b = encrypt(key, plaintext, associated_data)
return b[0] + b[1]
-def decrypt_pack(key: bytes, ciphertext: bytes):
+def decrypt_pack(key: AnyBytes, ciphertext: AnyBytes) -> bytes:
cp = memoryview(ciphertext)
return _decrypt(
key=key,
diff --git a/core/src/apps/monero/xmr/crypto_helpers.py b/core/src/apps/monero/xmr/crypto_helpers.py
index fe2fd0b2..9a350b65 100644
--- a/core/src/apps/monero/xmr/crypto_helpers.py
+++ b/core/src/apps/monero/xmr/crypto_helpers.py
@@ -12,26 +12,28 @@ from typing import TYPE_CHECKING
from trezor.crypto import monero as tcry
if TYPE_CHECKING:
+ from buffer_types import AnyBuffer, AnyBytes
+
from trezor.crypto.hashlib import sha3_256
NULL_KEY_ENC = b"\x00" * 32
-def get_keccak(data: bytes | None = None) -> sha3_256:
+def get_keccak(data: AnyBytes | None = None) -> sha3_256:
from trezor.crypto.hashlib import sha3_256
return sha3_256(data=data, keccak=True)
-def keccak_2hash(inp: bytes, buff: bytes | None = None) -> bytes:
+def keccak_2hash(inp: AnyBytes, buff: AnyBuffer | None = None) -> AnyBuffer:
buff = buff if buff else bytearray(32)
tcry.fast_hash_into(buff, inp)
tcry.fast_hash_into(buff, buff)
return buff
-def compute_hmac(key: bytes, msg: bytes) -> bytes:
+def compute_hmac(key: AnyBytes, msg: AnyBytes) -> bytes:
digestmod = get_keccak
inner = digestmod()
block_size = inner.block_size
@@ -59,7 +61,7 @@ def compute_hmac(key: bytes, msg: bytes) -> bytes:
#
-def decodepoint(x: bytes) -> tcry.Point:
+def decodepoint(x: AnyBytes) -> tcry.Point:
return tcry.decodepoint_into(None, x)
@@ -71,7 +73,7 @@ def encodeint(x: tcry.Scalar, offset: int = 0) -> bytes:
return tcry.encodeint_into(None, x, offset)
-def decodeint(x: bytes) -> tcry.Scalar:
+def decodeint(x: AnyBytes) -> tcry.Scalar:
return tcry.decodeint_into(None, x)
@@ -126,7 +128,7 @@ def get_subaddress_secret_key(
return tcry.xmr_get_subaddress_secret_key(None, major, minor, secret_key)
-def xor8(buff: bytearray, key: bytes) -> bytes:
+def xor8(buff: AnyBuffer, key: AnyBytes) -> AnyBytes:
for i in range(8):
buff[i] ^= key[i]
return buff
diff --git a/core/src/apps/monero/xmr/keccak_hasher.py b/core/src/apps/monero/xmr/keccak_hasher.py
index 93b156f7..6c2d078d 100644
--- a/core/src/apps/monero/xmr/keccak_hasher.py
+++ b/core/src/apps/monero/xmr/keccak_hasher.py
@@ -3,6 +3,8 @@ from typing import TYPE_CHECKING
from apps.monero.xmr.serialize import int_serialize
if TYPE_CHECKING:
+ from buffer_types import AnyBytes
+
from trezor.utils import HashContext, HashWriter
@@ -13,7 +15,7 @@ class KeccakXmrArchive:
def get_digest(self) -> bytes:
return self.kwriter.get_digest()
- def buffer(self, buf: bytes) -> None:
+ def buffer(self, buf: AnyBytes) -> None:
return self.kwriter.write(buf)
def uvarint(self, i: int) -> None:
diff --git a/core/src/apps/monero/xmr/key_image.py b/core/src/apps/monero/xmr/key_image.py
index 1bbddbd3..527f4b7f 100644
--- a/core/src/apps/monero/xmr/key_image.py
+++ b/core/src/apps/monero/xmr/key_image.py
@@ -3,12 +3,14 @@ from typing import TYPE_CHECKING
from apps.monero.xmr import crypto_helpers, monero
if TYPE_CHECKING:
+ from buffer_types import AnyBytes
+
from trezor.messages import MoneroTransferDetails
from apps.monero.xmr import crypto
from apps.monero.xmr.credentials import AccountCreds
- Subaddresses = dict[bytes, tuple[int, int]]
+ Subaddresses = dict[AnyBytes, tuple[int, int]]
Sig = list[list[crypto.Scalar]]
@@ -67,7 +69,7 @@ def export_key_image(
def generate_ring_signature(
- prefix_hash: bytes,
+ prefix_hash: AnyBytes,
image: crypto.Point,
pubs: list[crypto.Point],
sec: crypto.Scalar,
diff --git a/core/src/apps/monero/xmr/mlsag_hasher.py b/core/src/apps/monero/xmr/mlsag_hasher.py
index 1229d7c5..37a40a2c 100644
--- a/core/src/apps/monero/xmr/mlsag_hasher.py
+++ b/core/src/apps/monero/xmr/mlsag_hasher.py
@@ -1,9 +1,11 @@
from typing import TYPE_CHECKING
+from .serialize_messages.tx_rsig_bulletproof import BulletproofPlus
+
if TYPE_CHECKING:
- from trezor.utils import HashContext
+ from buffer_types import AnyBytes
- from .serialize_messages.tx_rsig_bulletproof import BulletproofPlus
+ from trezor.utils import HashContext
class PreMlsagHasher:
@@ -58,7 +60,7 @@ class PreMlsagHasher:
self.rtcsig_hasher = None # type: ignore
def rsig_val(
- self, p: bytes | list[bytes] | BulletproofPlus, raw: bool = False
+ self, p: AnyBytes | list[AnyBytes] | BulletproofPlus, raw: bool = False
) -> None:
if self.state == 8:
raise ValueError("State error")
@@ -73,11 +75,12 @@ class PreMlsagHasher:
for x in p:
update(x)
else:
- assert isinstance(p, bytes)
+ assert isinstance(p, (bytes, bytearray, memoryview))
update(p)
return
# Hash Bulletproof
+ assert isinstance(p, BulletproofPlus)
fields = (p.A, p.A1, p.B, p.r1, p.s1, p.d1)
for fld in fields:
update(fld)
diff --git a/core/src/apps/monero/xmr/monero.py b/core/src/apps/monero/xmr/monero.py
index 764bfe54..bcc01fc2 100644
--- a/core/src/apps/monero/xmr/monero.py
+++ b/core/src/apps/monero/xmr/monero.py
@@ -3,9 +3,11 @@ from typing import TYPE_CHECKING
from apps.monero.xmr import crypto, crypto_helpers
if TYPE_CHECKING:
+ from buffer_types import AnyBytes
+
from apps.monero.xmr.credentials import AccountCreds
- Subaddresses = dict[bytes, tuple[int, int]]
+ Subaddresses = dict[AnyBytes, tuple[int, int]]
class XmrException(Exception):
@@ -192,7 +194,7 @@ def generate_tx_spend_and_key_image_and_derivation(
out_key: crypto.Point,
tx_public_key: crypto.Point,
additional_tx_public_key: crypto.Point | None,
- real_output_index: int | None,
+ real_output_index: int,
sub_addr_major: int | None,
sub_addr_minor: int | None,
) -> tuple[crypto.Scalar, crypto.Point, crypto.Point]:
diff --git a/core/src/apps/monero/xmr/networks.py b/core/src/apps/monero/xmr/networks.py
index ea3d9181..d3a2940d 100644
--- a/core/src/apps/monero/xmr/networks.py
+++ b/core/src/apps/monero/xmr/networks.py
@@ -23,7 +23,7 @@ class StageNet:
def net_version(
- network_type: MoneroNetworkType = None,
+ network_type: MoneroNetworkType | None = None,
is_subaddr: bool = False,
is_integrated: bool = False,
) -> bytes:
Why this scored 18/100
Community notes
Notes can correct, qualify, or add evidence to the AI analysis. Every note shown here has been validated by a human moderator.
The AI analysis stands alone for now. Submit a note if you can add evidence or important context.