feat: Fix change-output classification to require verified descriptor derivation (#387)
What changed, and why it matters
This commit fixes a security bug in Specter DIY, a hardware wallet, where the device could wrongly hide a transaction output as 'change' (your own money coming back) when it actually went to an attacker or to another of your own wallet addresses. The old code trusted the host computer's claim that an output was change based only on which wallet owned the inputs. The new code independently re-derives the output address from the wallet's descriptor and only hides it as change when the math matches exactly, the descriptor has the standard two branches, and the output is on the change branch. It also warns the user when the host's metadata does not match.
Users should upgrade to a firmware build containing this commit. After upgrading, verify that transaction confirmation screens still show all non-change outputs and any outputs with warnings. Developers should review the new test suite for completeness and ensure the descriptor cache does not introduce side channels on the constrained hardware target.
Security signals we found
Fixes incorrect change-output classification that could hide attacker-controlled or self-payment outputs
Adds cryptographic verification of PSBT derivation metadata against locally stored descriptor and output scriptPubKey
Adds explicit warnings for inconsistent host-supplied change metadata
Narrows automatic change classification to two-branch descriptors with branch-list position 1
Adds extensive regression tests covering forged metadata, multi-branch descriptors, Taproot, Liquid, and UI visibility
Updates security model documentation to describe the new conservative behavior
Evidence from the diff
The patch replaces a naive change-output classification rule in WalletManager with a cryptographic verification. Previously, an output was marked change if wallet is not None and len(wallets) == 1 and wallet in wallets. Now get_output_status() checks each PSBT BIP32/Taproot derivation claim against the stored descriptor by re-deriving the public key and scriptPubKey. Change classification requires: exactly one spending wallet, the output’s wallet is that wallet, the descriptor has exactly two branches, the claim resolves to branch-list position 1, and the re-derived scriptPubKey equals the actual output script. Forged branch-1 metadata that does not match the script triggers an ‘Invalid change metadata!’ warning and keeps the output visible. Outputs with unknown inputs or non-standard descriptors are treated conservatively as visible outputs. The UI was updated to skip only verified, warning-free change outputs on the primary confirmation screen, while the details page still lists every output.
Changed components
src/apps/wallets/manager.pysrc/apps/wallets/liquid/manager.pysrc/gui/screens/transaction.pydocs/descriptors.mddocs/security-model.mdInspect captured patch +1514 / −59
### docs/descriptors.md
@@ -17,6 +17,15 @@ Here for receiving address number 17 the wallet will use the script from `wsh(so
The only requirement is that the number of indexes in all sets is the same (3 in the case above).
+Automatic change classification is deliberately narrower than descriptor
+parsing. The device treats only descriptors with exactly two branches as
+having the canonical receive/change shape. In a two-branch descriptor,
+branch-list position 0 is receive and position 1 is change. A raw child
+number is not the position: in `<22;33>`, raw value 33 is position 1 and is
+eligible for change classification. Descriptors with one, three, or more
+branches remain valid descriptors, but their outputs are never hidden as
+automatic change.
+
## Default derivations
If the descriptor contains master public keys but doesn't contain wildcard derivations, the default derivation `/{0,1}/*` will be added to all extended keys in the descriptor. If at least one of the xpubs has a wildcard derivation the descriptor will not be changed.
### docs/security-model.md
@@ -306,10 +306,25 @@ The following rules apply to transactions that the wallet will sign:
wallet by adding the wallet descriptor (over QR, USB or SD card). The
device only signs for wallets it knows.
-Change is verified for you automatically: the device identifies change
-outputs against the imported wallet descriptor and labels them with the
-wallet name. What the device cannot check is the *recipient* — so always
-verify the receive address and the transaction details (amounts, fees)
+Change is verified for you automatically only when there is one unambiguous
+spending wallet, its descriptor has exactly two branches, the output
+derivation resolves to branch-list position 1, and the device re-derives the
+exact output script from that branch and index. Every other output remains
+visible, including receive-branch self-payments and outputs from unusual
+descriptors. The branch position is not the raw child-number value in the
+descriptor.
+
+The Bitcoin transaction does not contain an `is_change` flag. A PSBT may carry
+host-supplied BIP32 or Taproot derivation metadata, but that metadata is
+untrusted and is checked against the locally stored descriptor and the actual
+output script. If a branch-1 wallet claim does not match, the output remains
+visible and the device displays: `Invalid change metadata! Host claimed this
+output as wallet change, but it does not match your wallet. Verify the
+destination.` This reports inconsistent host metadata; it does not claim that
+Bitcoin marked the output as change.
+
+What the device cannot check is the *recipient* of an unverified output — so
+always verify the receive address and the transaction details (amounts, fees)
on the device screen. The screen is the trusted output channel, the host
computer is not.
### src/apps/wallets/liquid/manager.py
@@ -338,7 +338,9 @@ def preprocess_psbt(self, stream, fout):
gaps = None
if wallet:
gaps = [g for g in wallet.gaps] # copy
- res = wallet.get_derivation(inp.bip32_derivations)
+ res = wallet.get_derivation(
+ inp.bip32_derivations, getattr(inp, "taproot_bip32_derivations", {})
+ )
if res:
idx, branch_idx = res
gaps[branch_idx] = max(gaps[branch_idx], idx+wallet.GAP_LIMIT+1)
@@ -543,30 +545,10 @@ def preprocess_psbt(self, stream, fout):
asset = None
value = -1
metaout.update({
- "change": (wallet is not None and len(wallets) == 1 and wallet in wallets),
"value": value,
- "address": self.get_address(out),
"asset": self.asset_label(asset),
})
- if wallet:
- metaout["label"] = wallet.name
- res = wallet.get_derivation(out.bip32_derivations)
- if res:
- idx, branch_idx = res
- branch_txt = ""
- if branch_idx == 1:
- "change "
- elif branch_idx > 1:
- "branch %d " % branch_idx
- metaout["label"] = "%s %s#%d" % (wallet.name, branch_txt, idx)
- if wallet in wallets:
- allowed_idx = wallets[wallet]["gaps"][branch_idx]
- else:
- allowed_idx = wallet.gaps[branch_idx]
- if allowed_idx <= idx:
- metaout["warning"] = "Derivation index is by %d larger than last known used index %d!" % (idx-allowed_idx+wallet.GAP_LIMIT, allowed_idx-wallet.GAP_LIMIT)
- if wallet.is_watchonly:
- metaout["warning"] = "Watch-only wallet!"
+ self.fill_output_metadata(metaout, wallet, wallets, out)
if asset and asset not in self.assets:
metaout.update({"raw_asset": asset})
out.write_to(fout, skip_separator=True, version=psbtv.version)
### src/apps/wallets/manager.py
@@ -43,6 +43,17 @@
for sh in list(SIGHASH_NAMES):
SIGHASH_NAMES[sh | SIGHASH.ANYONECANPAY] = SIGHASH_NAMES[sh] + " | ANYONECANPAY"
+INVALID_CHANGE_METADATA_WARNING = (
+ "Invalid change metadata! Host claimed this output as wallet change, "
+ "but it does not match your wallet. Verify the destination."
+)
+UNVERIFIED_CHANGE_WARNING = (
+ "This output goes to your wallet's change address (branch %d, #%d), "
+ "but the transaction contains unknown inputs. It is being treated as a "
+ "regular wallet output because it cannot be verified as change. "
+ "Review the destination."
+)
+
class WalletManager(BaseApp):
"""
WalletManager class manages your wallets.
@@ -385,6 +396,231 @@ async def confirm_wallets(self, wallets, show_screen):
proceed = await show_screen(scr)
return proceed
+ def add_output_warning(self, metaout, warning):
+ """Append an output warning without discarding an earlier warning."""
+ if not warning:
+ return
+ warnings = metaout.setdefault("warnings", [])
+ if warning not in warnings:
+ warnings.append(warning)
+
+ def _get_key_derivation_claims(
+ self, wallet, pubkey, derivation, is_taproot, descriptor_cache
+ ):
+ """Return ``(claim, descriptor)`` pairs matching a PSBT key and path."""
+ claims = []
+ for key_idx, key in enumerate(wallet.descriptor.keys):
+ try:
+ claim = key.check_derivation(derivation)
+ if claim is None:
+ continue
+ idx, branch_idx = claim
+ cache_key = (wallet, idx, branch_idx)
+ desc = descriptor_cache.get(cache_key)
+ if desc is None:
+ desc, _ = wallet.get_descriptor(idx, branch_idx)
+ descriptor_cache[cache_key] = desc
+ derived_pubkey = desc.keys[key_idx].get_public_key()
+ if is_taproot:
+ key_matches = derived_pubkey.xonly() == pubkey.xonly()
+ # BIP 371 also permits the tweaked output key.
+ if not key_matches and key is wallet.descriptor.key:
+ key_matches = (
+ desc.script_pubkey().data[2:] == pubkey.xonly()
+ )
+ else:
+ key_matches = derived_pubkey == pubkey
+ if not key_matches:
+ continue
+ duplicate = False
+ for existing_claim, _ in claims:
+ if existing_claim == claim:
+ duplicate = True
+ break
+ if not duplicate:
+ claims.append((claim, desc))
+ except Exception:
+ continue
+ return claims
+
+ def get_wallet_derivation_claims(self, wallet, out, output_wallet=None):
+ """Return derivation claims attributable to ``wallet`` for an output."""
+ claims = []
+ descriptor_cache = {}
+ derivation_sets = (
+ (False, getattr(out, "bip32_derivations", {}).items()),
+ (True, ((pubkey, derivation) for pubkey, (leafs, derivation) in
+ getattr(out, "taproot_bip32_derivations", {}).items())),
+ )
+ for is_taproot, derivations in derivation_sets:
+ for pubkey, derivation in derivations:
+ matching_claims = self._get_key_derivation_claims(
+ wallet, pubkey, derivation, is_taproot, descriptor_cache
+ )
+ try:
+ claim = wallet.descriptor.check_derivation(derivation)
+ except Exception:
+ claim = None
+ # Different descriptors can legitimately reuse the same key
+ # origins and derivation paths. If this metadata also derives
+ # the actual output under its known destination wallet, it is
+ # not an exclusive change claim from the spending wallet.
+ if (
+ claim is not None
+ and output_wallet is not None
+ and output_wallet is not wallet
+ ):
+ try:
+ output_claims = self._get_key_derivation_claims(
+ output_wallet, pubkey, derivation, is_taproot,
+ descriptor_cache
+ )
+ if any(
+ out.script_pubkey is not None
+ and desc.script_pubkey() == out.script_pubkey
+ for _, desc in output_claims
+ ):
+ matching_claims = []
+ claim = None
+ except Exception:
+ pass
+ if matching_claims:
+ output_matches = [
+ matching_claim
+ for matching_claim, desc in matching_claims
+ if out.script_pubkey is not None
+ and desc.script_pubkey() == out.script_pubkey
+ ]
+ entry_claims = output_matches or [
+ matching_claim
+ for matching_claim, desc in matching_claims
+ ]
+ else:
+ # Keep malformed path/pubkey metadata visible as suspicious.
+ entry_claims = [] if claim is None else [claim]
+ for entry_claim in entry_claims:
+ if entry_claim not in claims:
+ claims.append(entry_claim)
+ return claims
+
+ def get_output_status(self, wallet, wallets, out):
+ """Return ``(derivation, is_change, warning)`` for one output.
+
+ Output ownership must not erase a descriptor-valid claim from the sole
+ known spending wallet. Claims from unrelated imported wallets and
+ ambiguous transactions are deliberately ignored.
+ """
+ derivation = None
+ if wallet is not None:
+ for claim in self.get_wallet_derivation_claims(wallet, out):
+ try:
+ desc, _ = wallet.get_descriptor(*claim)
+ if desc.script_pubkey() == out.script_pubkey:
+ derivation = claim
+ break
+ except Exception:
+ continue
+ if derivation is None:
+ derivation = wallet.get_derivation(
+ out.bip32_derivations,
+ getattr(out, "taproot_bip32_derivations", {}),
+ )
+ warning = None
+ spending_wallets = [w for w in wallets if w is not None]
+ # Only a two-branch descriptor gives branch-list position 1 the
+ # "change" meaning. For <0;1;2> and other unusual layouts we do
+ # not know what position 1 is, so a host derivation for it is not
+ # a change claim and must not raise INVALID_CHANGE_METADATA - same
+ # rule the is_change classification below uses.
+ if (
+ len(spending_wallets) == 1
+ and spending_wallets[0].descriptor.num_branches == 2
+ ):
+ candidate = spending_wallets[0]
+ branch1_claims = [
+ claim for claim in self.get_wallet_derivation_claims(
+ candidate, out, output_wallet=wallet
+ )
+ if claim[1] == 1
+ ]
+ for idx, branch_idx in branch1_claims:
+ try:
+ desc, _ = candidate.get_descriptor(idx, branch_idx)
+ except Exception:
+ continue
+ if out.script_pubkey is not None and desc.script_pubkey() == out.script_pubkey:
+ break
+ else:
+ if branch1_claims:
+ warning = INVALID_CHANGE_METADATA_WARNING
+
+ is_change = False
+ if wallet is not None and derivation is not None:
+ idx, branch_idx = derivation
+ if (
+ len(wallets) == 1
+ and wallet in wallets
+ and wallet.descriptor.num_branches == 2
+ and branch_idx == 1
+ ):
+ try:
+ desc, _ = wallet.get_descriptor(idx, branch_idx)
+ is_change = desc.script_pubkey() == out.script_pubkey
+ except Exception:
+ is_change = False
+ if (
+ not is_change
+ and None in wallets
+ and wallet.descriptor.num_branches == 2
+ and branch_idx == 1
+ ):
+ try:
+ desc, _ = wallet.get_descriptor(idx, branch_idx)
+ if warning is None and desc.script_pubkey() == out.script_pubkey:
+ warning = UNVERIFIED_CHANGE_WARNING % (branch_idx, idx)
+ except Exception:
+ pass
+ return derivation if wallet is not None else None, is_change, warning
+
+ def fill_output_metadata(self, metaout, wallet, wallets, out):
+ """Fill in the wallet-related output metadata shared by all networks.
+
+ Sets the change flag, address, label and warnings. The value and any
+ network-specific fields stay with the caller.
+ """
+ derivation, is_change, warning = self.get_output_status(wallet, wallets, out)
+ metaout.update({
+ "change": is_change,
+ "address": self.get_address(out),
+ })
+ if warning:
+ self.add_output_warning(metaout, warning)
+ if not wallet:
+ return
+ metaout["label"] = wallet.name
+ if derivation:
+ idx, branch_idx = derivation
+ if is_change:
+ metaout["label"] = "%s change #%d" % (wallet.name, idx)
+ else:
+ # Label by is_change, never by branch_idx alone, so an output
+ # the security logic did not accept as change can't still be
+ # captioned "change".
+ branch_txt = "" if branch_idx == 0 else "branch %d " % branch_idx
+ metaout["label"] = "This wallet (%s) %s#%d" % (wallet.name, branch_txt, idx)
+ if wallet in wallets:
+ allowed_idx = wallets[wallet]["gaps"][branch_idx]
+ else:
+ allowed_idx = wallet.gaps[branch_idx]
+ if allowed_idx <= idx:
+ self.add_output_warning(
+ metaout,
+ "Derivation index is by %d larger than last known used index %d!" %
+ (idx-allowed_idx+wallet.GAP_LIMIT, allowed_idx-wallet.GAP_LIMIT),
+ )
+ if wallet.is_watchonly:
+ self.add_output_warning(metaout, "Watch-only wallet!")
+
def get_sighash_info(self, sighash):
if sighash not in SIGHASH_NAMES:
raise WalletError("Unknown sighash type: %d!" % sighash)
@@ -683,7 +919,9 @@ def preprocess_psbt(self, stream, fout):
break
if wallet:
gaps = [g for g in wallet.gaps] # copy
- res = wallet.get_derivation(inp.bip32_derivations)
+ res = wallet.get_derivation(
+ inp.bip32_derivations, getattr(inp, "taproot_bip32_derivations", {})
+ )
if res:
idx, branch_idx = res
gaps[branch_idx] = max(gaps[branch_idx], idx+wallet.GAP_LIMIT+1)
@@ -740,30 +978,8 @@ def preprocess_psbt(self, stream, fout):
# Get values and store in metadata and wallets dict
value = out.value
fee -= value
- metaout.update({
- "change": (wallet is not None and len(wallets) == 1 and wallet in wallets),
- "value": value,
- "address": self.get_address(out),
- })
- if wallet:
- metaout["label"] = wallet.name
- res = wallet.get_derivation(out.bip32_derivations)
- if res:
- idx, branch_idx = res
- branch_txt = ""
- if branch_idx == 1:
- "change "
- elif branch_idx > 1:
- "branch %d " % branch_idx
- metaout["label"] = "%s %s#%d" % (wallet.name, branch_txt, idx)
- if wallet in wallets:
- allowed_idx = wallets[wallet]["gaps"][branch_idx]
- else:
- allowed_idx = wallet.gaps[branch_idx]
- if allowed_idx <= idx:
- metaout["warning"] = "Derivation index is by %d larger than last known used index %d!" % (idx-allowed_idx+wallet.GAP_LIMIT, allowed_idx-wallet.GAP_LIMIT)
- if wallet.is_watchonly:
- metaout["warning"] = "Watch-only wallet!"
+ metaout["value"] = value
+ self.fill_output_metadata(metaout, wallet, wallets, out)
out.write_to(fout, version=psbtv.version)
meta["fee"] = fee
### src/gui/screens/transaction.py
@@ -71,11 +71,12 @@ def __init__(self, title, meta):
self.warning = self.add_warning(self.page, warning_text)
obj = self.warning
- num_change_outputs = 0
for out in meta["outputs"]:
- # first only show destination addresses
- if out["change"] and not out.get("warning", ""):
- num_change_outputs += 1
+ # Verified change needs no confirmation - the device proved it
+ # can't be an attacker-controlled destination. It stays visible
+ # on the details page. A warning overrides this: it means the
+ # output needs the user's attention.
+ if out["change"] and not out.get("warnings"):
continue
obj = self.show_output(out, obj)
@@ -183,8 +184,9 @@ def __init__(self, title, meta):
else:
addrlbl.set_style(0, style_primary)
lbl = addrlbl
- if "warning" in out:
- text = out["warning"]
+ warning_text = "\n".join(out.get("warnings", []))
+ if warning_text:
+ text = warning_text
warning = add_label(text, scr=self.page2)
warning.set_align(lv.label.ALIGN.LEFT)
warning.set_width(380)
@@ -274,8 +276,9 @@ def show_output(self, out, obj):
addr.set_style(0, self.style)
addr.align(obj, lv.ALIGN.OUT_BOTTOM_MID, 0, 10)
obj = addr
- if "warning" in out:
- text = "WARNING! %s" % out["warning"]
+ warning_text = "\n".join(out.get("warnings", []))
+ if warning_text:
+ text = "WARNING! %s" % warning_text
warning = add_label(text, scr=self.page)
warning.set_style(0, self.style_warning)
warning.align(obj, lv.ALIGN.OUT_BOTTOM_MID, 0, 10)
### test/tests_native/__init__.py
@@ -1,3 +1,6 @@
from .test_manifest_inventory import *
from .test_wallet_manager_parsing import *
from .test_wallet_manager_warnings import *
+from .test_change_classification import *
+from .test_transaction_confirmation import *
+from .test_change_security import *
### test/tests_native/test_change_classification.py
@@ -0,0 +1,471 @@
+import sys
+
+if sys.implementation.name != 'micropython':
+ from native_support import setup_native_stubs
+
+ setup_native_stubs()
+
+from io import BytesIO
+from types import SimpleNamespace
+from unittest import TestCase
+import gc
+
+from tests.util import get_keystore, get_wallets_app, clear_testdir
+
+from embit import bip32, ec, script
+from embit.psbt import PSBT, DerivationPath
+from embit.transaction import Transaction, TransactionInput, TransactionOutput
+
+from apps.wallets.wallet import Wallet
+
+# Regression coverage for the change-output classification fix.
+#
+# Historically an output was classified as change purely because it
+# belonged to the single wallet that also owned the transaction inputs:
+#
+# "change": (wallet is not None and len(wallets) == 1 and wallet in wallets)
+#
+# That does not distinguish a receive-branch (self-payment) output from a
+# real change output, and it trusts host-supplied BIP32 derivation metadata
+# without ever checking it against the actual output script_pubkey. These
+# tests exercise WalletManager.get_output_status() - the
+# function that replaced that logic - directly, plus one full
+# preprocess_psbt() pipeline test that reproduces the exact adversarial
+# scenario described in the security report.
+
+
+def fake_pubkey(seed):
+ return ec.PrivateKey(bytes([seed]) * 32).get_public_key()
+
+
+class ChangeClassificationTest(TestCase):
+ def setUp(self):
+ clear_testdir()
+ self.keystore = get_keystore()
+ self.wallets_app = get_wallets_app(self.keystore, "regtest")
+ self.manager = self.wallets_app.manager
+ self.wallet = self.manager.wallets[0] # default wpkh([..]/{0,1}/*) wallet
+ self.fingerprint = self.keystore.fingerprint
+
+ def tearDown(self):
+ clear_testdir()
+ gc.collect()
+
+ def derivation(self, wallet, branch_idx, idx, origin="m/84h/1h/0h"):
+ path = bip32.parse_path("%s/%d/%d" % (origin, branch_idx, idx))
+ return DerivationPath(self.fingerprint, path)
+
+ def script_for(self, wallet, branch_idx, idx):
+ return wallet.descriptor.derive(idx, branch_index=branch_idx).script_pubkey()
+
+ def make_out(self, bip32_derivations=None, taproot_bip32_derivations=None, script_pubkey=None):
+ return SimpleNamespace(
+ bip32_derivations=bip32_derivations or {},
+ taproot_bip32_derivations=taproot_bip32_derivations or {},
+ script_pubkey=script_pubkey,
+ )
+
+ # --- Test A: legitimate verified change -------------------------------
+
+ def test_verified_change_branch1_is_change(self):
+ out = self.make_out(
+ bip32_derivations={fake_pubkey(1): self.derivation(self.wallet, 1, 3)},
+ script_pubkey=self.script_for(self.wallet, 1, 3),
+ )
+ wallets = {self.wallet: {}}
+ derivation, is_change, _ = self.manager.get_output_status(
+ self.wallet, wallets, out
+ )
+ self.assertEqual(derivation, (3, 1))
+ self.assertTrue(is_change)
+
+ # --- Test B: receive-branch (self-payment) output of the spending wallet
+ # This is the critical regression case: a genuine output of the sole
+ # spending wallet, on branch 0, must NOT be treated as change. This
+ # assertion fails against the previous implementation, which only
+ # checked `wallet is not None and len(wallets) == 1`.
+
+ def test_receive_branch_output_is_not_change(self):
+ out = self.make_out(
+ bip32_derivations={fake_pubkey(2): self.derivation(self.wallet, 0, 5)},
+ script_pubkey=self.script_for(self.wallet, 0, 5),
+ )
+ wallets = {self.wallet: {}}
+ derivation, is_change, _ = self.manager.get_output_status(
+ self.wallet, wallets, out
+ )
+ self.assertEqual(derivation, (5, 0))
+ self.assertFalse(is_change)
+
+ # --- Test C: forged change derivation - metadata claims branch 1, but the
+ # actual output script does not match what the device derives for it.
+
+ def test_forged_branch1_metadata_with_mismatched_script_is_not_change(self):
+ wrong_script = self.script_for(self.wallet, 0, 3) # real script, wrong branch
+ out = self.make_out(
+ bip32_derivations={fake_pubkey(3): self.derivation(self.wallet, 1, 3)},
+ script_pubkey=wrong_script,
+ )
+ wallets = {self.wallet: {}}
+ derivation, is_change, _ = self.manager.get_output_status(
+ self.wallet, wallets, out
+ )
+ self.assertFalse(is_change)
+
+ def test_forged_script_pubkey_unrelated_to_wallet_is_not_change(self):
+ # metadata claims branch 1 idx 3, but script_pubkey is a completely
+ # unrelated (attacker controlled) script
+ out = self.make_out(
+ bip32_derivations={fake_pubkey(3): self.derivation(self.wallet, 1, 3)},
+ script_pubkey=script.p2wpkh(fake_pubkey(99)),
+ )
+ wallets = {self.wallet: {}}
+ derivation, is_change, _ = self.manager.get_output_status(
+ self.wallet, wallets, out
+ )
+ self.assertFalse(is_change)
+
+ # --- Test D: output that doesn't belong to any known wallet ------------
+
+ def test_unknown_wallet_output_is_not_change(self):
+ out = self.make_out(script_pubkey=script.p2wpkh(fake_pubkey(42)))
+ wallets = {self.wallet: {}}
+ derivation, is_change, _ = self.manager.get_output_status(
+ None, wallets, out
+ )
+ self.assertIsNone(derivation)
+ self.assertFalse(is_change)
+
+ # --- Test E: multiple spending wallets - conservative behaviour must be
+ # preserved even for a verified branch-1 output.
+
+ def test_verified_change_with_multiple_spending_wallets_is_not_change(self):
+ other_wallet = Wallet.from_descriptor(str(self.wallet.descriptor), None)
+ out = self.make_out(
+ bip32_derivations={fake_pubkey(4): self.derivation(self.wallet, 1, 3)},
+ script_pubkey=self.script_for(self.wallet, 1, 3),
+ )
+ wallets = {self.wallet: {}, other_wallet: {}}
+ derivation, is_change, _ = self.manager.get_output_status(
+ self.wallet, wallets, out
+ )
+ self.assertEqual(derivation, (3, 1))
+ self.assertFalse(is_change)
+
+ def test_output_wallet_not_among_spending_wallets_is_not_change(self):
+ # wallet owns the output (verified), but it never appeared as an
+ # input-owning wallet in this transaction
+ out = self.make_out(
+ bip32_derivations={fake_pubkey(5): self.derivation(self.wallet, 1, 3)},
+ script_pubkey=self.script_for(self.wallet, 1, 3),
+ )
+ wallets = {} # no wallet owns any input
+ derivation, is_change, _ = self.manager.get_output_status(
+ self.wallet, wallets, out
+ )
+ self.assertFalse(is_change)
+
+ # --- Test F: additional descriptor branch (branch index > 1) -----------
+
+ def test_branch_beyond_change_is_not_change(self):
+ der_path = "m/84h/1h/0h"
+ xpub = self.keystore.get_xpub(der_path)
+ desc_str = "wpkh([%s%s]%s/<0;1;2>/*)" % (
+ self.fingerprint.hex(),
+ der_path[1:],
+ xpub.to_base58(self.manager.Networks["regtest"]["xpub"]),
+ )
+ wallet3 = Wallet.from_descriptor(desc_str, None)
+ self.assertEqual(wallet3.descriptor.num_branches, 3)
+
+ out = self.make_out(
+ bip32_derivations={fake_pubkey(6): self.derivation(wallet3, 2, 7)},
+ script_pubkey=self.script_for(wallet3, 2, 7),
+ )
+ wallets = {wallet3: {}}
+ derivation, is_change, _ = self.manager.get_output_status(
+ wallet3, wallets, out
+ )
+ self.assertEqual(derivation, (7, 2))
+ self.assertFalse(is_change)
+
+ def test_branch_index_is_position_not_raw_derivation_value(self):
+ # branch_idx must mean "position in the descriptor's branch list",
+ # not the raw derivation value at that path component. A descriptor
+ # using unusual branch values (22, 33, 44) still has its change
+ # branch at position 1 (raw value 33) - not at raw value 1, which
+ # isn't even a valid branch of this descriptor.
+ der_path = "m/84h/1h/0h"
+ xpub = self.keystore.get_xpub(der_path)
+ desc_str = "wpkh([%s%s]%s/<22;33;44>/*)" % (
+ self.fingerprint.hex(),
+ der_path[1:],
+ xpub.to_base58(self.manager.Networks["regtest"]["xpub"]),
+ )
+ odd_wallet = Wallet.from_descriptor(desc_str, None)
+ self.assertEqual(odd_wallet.descriptor.num_branches, 3)
+
+ def odd_derivation(raw_branch_value, idx):
+ path = bip32.parse_path("%s/%d/%d" % (der_path, raw_branch_value, idx))
+ return DerivationPath(self.fingerprint, path)
+
+ wallets = {odd_wallet: {}}
+
+ # raw derivation value 33 -> branch position 1, but the descriptor
+ # has three branches so it is not eligible for automatic change.
+ out_change = self.make_out(
+ bip32_derivations={fake_pubkey(21): odd_derivation(33, 4)},
+ script_pubkey=self.script_for(odd_wallet, 1, 4),
+ )
+ derivation, is_change, _ = self.manager.get_output_status(
+ odd_wallet, wallets, out_change
+ )
+ self.assertEqual(derivation, (4, 1))
+ self.assertFalse(is_change)
+
+ # raw derivation value 22 -> branch position 0 -> receive, not change
+ out_receive = self.make_out(
+ bip32_derivations={fake_pubkey(22): odd_derivation(22, 4)},
+ script_pubkey=self.script_for(odd_wallet, 0, 4),
+ )
+ derivation0, is_change0, _ = self.manager.get_output_status(
+ odd_wallet, wallets, out_receive
+ )
+ self.assertEqual(derivation0, (4, 0))
+ self.assertFalse(is_change0)
+
+ # --- Test H: Taproot ----------------------------------------------------
+
+ def test_taproot_branch1_is_change_and_branch0_is_not(self):
+ der_path = "m/86h/1h/0h"
+ xpub = self.keystore.get_xpub(der_path)
+ desc_str = "tr([%s%s]%s/<0;1>/*)" % (
+ self.fingerprint.hex(),
+ der_path[1:],
+ xpub.to_base58(self.manager.Networks["regtest"]["xpub"]),
+ )
+ tr_wallet = Wallet.from_descriptor(desc_str, None)
+ self.assertTrue(tr_wallet.descriptor.is_taproot)
+ wallets = {tr_wallet: {}}
+
+ change_der = self.derivation(tr_wallet, 1, 3, origin=der_path)
+ out_change = self.make_out(
+ taproot_bip32_derivations={fake_pubkey(7): ([], change_der)},
+ script_pubkey=self.script_for(tr_wallet, 1, 3),
+ )
+ derivation, is_change, _ = self.manager.get_output_status(
+ tr_wallet, wallets, out_change
+ )
+ self.assertEqual(derivation, (3, 1))
+ self.assertTrue(is_change)
+
+ recv_der = self.derivation(tr_wallet, 0, 3, origin=der_path)
+ out_recv = self.make_out(
+ taproot_bip32_derivations={fake_pubkey(8): ([], recv_der)},
+ script_pubkey=self.script_for(tr_wallet, 0, 3),
+ )
+ derivation0, is_change0, _ = self.manager.get_output_status(
+ tr_wallet, wallets, out_recv
+ )
+ self.assertEqual(derivation0, (3, 0))
+ self.assertFalse(is_change0)
+
+ # --- Test I: Liquid -------------------------------------------------
+
+ def test_liquid_branch0_not_change_branch1_is_change(self):
+ clear_testdir()
+ lks = get_keystore()
+ lwapp = get_wallets_app(lks, "elementsregtest")
+ lmanager = lwapp.manager
+ lwallet = lmanager.wallets[0]
+ lfp = lks.fingerprint
+
+ def lderivation(branch_idx, idx):
+ path = bip32.parse_path("m/84h/1h/0h/%d/%d" % (branch_idx, idx))
+ return DerivationPath(lfp, path)
+
+ wallets = {lwallet: {}}
+
+ out_recv = self.make_out(
+ bip32_derivations={fake_pubkey(11): lderivation(0, 5)},
+ script_pubkey=lwallet.descriptor.derive(5, branch_index=0).script_pubkey(),
+ )
+ derivation, is_change, _ = lmanager.get_output_status(
+ lwallet, wallets, out_recv
+ )
+ self.assertEqual(derivation, (5, 0))
+ self.assertFalse(is_change)
+
+ out_change = self.make_out(
+ bip32_derivations={fake_pubkey(12): lderivation(1, 3)},
+ script_pubkey=lwallet.descriptor.derive(3, branch_index=1).script_pubkey(),
+ )
+ derivation2, is_change2, _ = lmanager.get_output_status(
+ lwallet, wallets, out_change
+ )
+ self.assertEqual(derivation2, (3, 1))
+ self.assertTrue(is_change2)
+ clear_testdir()
+
+ # --- Adversarial regression transaction (full preprocess_psbt pipeline) -
+
+ def test_adversarial_regression_transaction(self):
+ """
+ Reproduces the exact scenario from the security report end-to-end
+ through WalletManager.preprocess_psbt():
+
+ - Input: Wallet A
+ - Output 0: external recipient -> not change, visible
+ - Output 1: Wallet A receive branch (/0/5) -> not change, visible
+ - Output 2: Wallet A verified change branch (/1/3) -> change, visible
+
+ Against the pre-fix implementation, output 1 was misclassified as
+ change (since Wallet A is the sole spending wallet), which could
+ make it disappear from the primary confirmation screen. This test
+ fails on the old implementation and passes after the fix.
+ """
+ wallet = self.wallet
+ prev_script = self.script_for(wallet, 0, 0)
+ txin = TransactionInput(b"\x11" * 32, 0)
+
+ external_script = script.p2wpkh(fake_pubkey(50))
+ out0 = TransactionOutput(50_000, external_script)
+ out1 = TransactionOutput(100_000, self.script_for(wallet, 0, 5))
+ out2 = TransactionOutput(40_000, self.script_for(wallet, 1, 3))
+
+ tx = Transaction(vin=[txin], vout=[out0, out1, out2])
+ p = PSBT(tx)
+ p.inputs[0].witness_utxo = TransactionOutput(190_000, prev_script)
+ p.inputs[0].bip32_derivations[fake_pubkey(51)] = self.derivation(wallet, 0, 0)
+ p.outputs[1].bip32_derivations[fake_pubkey(52)] = self.derivation(wallet, 0, 5)
+ p.outputs[2].bip32_derivations[fake_pubkey(53)] = self.derivation(wallet, 1, 3)
+
+ raw = p.serialize()
+ fout = BytesIO()
+ wallets, meta = self.manager.preprocess_psbt(BytesIO(raw), fout)
+
+ outputs = meta["outputs"]
+ self.assertEqual(len(outputs), 3)
+
+ # Output 0: external, unknown wallet
+ self.assertFalse(outputs[0]["change"])
+ self.assertNotIn("label", outputs[0])
+
+ # Output 1: receive-branch self-payment - must NOT be change
+ self.assertFalse(outputs[1]["change"])
+ self.assertIn("label", outputs[1])
+ self.assertNotIn("change", outputs[1]["label"])
+
+ # Output 2: verified change on branch 1
+ self.assertTrue(outputs[2]["change"])
+ self.assertIn("change", outputs[2]["label"])
+
+ # send_amount (as computed in the GUI) must include outputs 0 and 1
+ # but exclude the verified change output.
+ send_amount = sum(out["value"] for out in outputs if not out["change"])
+ self.assertEqual(send_amount, 50_000 + 100_000)
+
+ # --- Label must follow is_change, never branch_idx alone ---------------
+ #
+ # preprocess_psbt() used to choose the output label from branch_idx
+ # directly ("... change #idx" whenever branch_idx == 1), instead of
+ # from the is_change verdict that get_output_status()
+ # already computed. That means an output correctly NOT treated as
+ # change (still fully shown to the user) could still be captioned
+ # "change" - misleading, since the trusted display would be telling
+ # the user "change" while its own security logic says otherwise. Both
+ # cases below hit a real branch-1 derivation that is_change rejects
+ # for a reason other than "wrong branch".
+
+ def _other_wallet(self, mnemonic, name="WalletB"):
+ # a second, independently keyed wallet - imported like a real
+ # watch-only wallet would be, not a clone of self.wallet's keys.
+ # Returns (wallet, fingerprint): self.derivation() always signs
+ # with self.fingerprint (Wallet A's), so derivations for this
+ # wallet must be built with its own fingerprint instead.
+ other_keystore = get_keystore(mnemonic=mnemonic)
+ der_path = "m/84h/1h/0h"
+ xpub = other_keystore.get_xpub(der_path)
+ desc_str = "wpkh([%s%s]%s/<0;1>/*)" % (
+ other_keystore.fingerprint.hex(),
+ der_path[1:],
+ xpub.to_base58(self.manager.Networks["regtest"]["xpub"]),
+ )
+ wallet = Wallet.from_descriptor(desc_str, None)
+ wallet.name = name
+ self.manager.wallets.append(wallet)
+ return wallet, other_keystore.fingerprint
+
+ def _derivation_for(self, fingerprint, branch_idx, idx, origin="m/84h/1h/0h"):
+ path = bip32.parse_path("%s/%d/%d" % (origin, branch_idx, idx))
+ return DerivationPath(fingerprint, path)
+
+ def test_branch1_output_of_a_different_wallet_is_not_labeled_change(self):
+ # Wallet A spends the only input. The output verifiably belongs to
+ # Wallet B (also imported, but not among this transaction's
+ # spending wallets) on branch 1. get_output_status()
+ # correctly returns is_change=False (Wallet B isn't a spending
+ # wallet here), so the output stays fully visible - but its label
+ # must not claim "change" for it.
+ wallet_b, wallet_b_fp = self._other_wallet("zoo " * 11 + "wrong")
+
+ wallet_a = self.wallet
+ prev_script = self.script_for(wallet_a, 0, 0)
+ txin = TransactionInput(b"\x22" * 32, 0)
+ out0 = TransactionOutput(
+ 29_000, wallet_b.descriptor.derive(5, branch_index=1).script_pubkey()
+ )
+
+ tx = Transaction(vin=[txin], vout=[out0])
+ p = PSBT(tx)
+ p.inputs[0].witness_utxo = TransactionOutput(30_000, prev_script)
+ p.inputs[0].bip32_derivations[fake_pubkey(61)] = self.derivation(wallet_a, 0, 0)
+ p.outputs[0].bip32_derivations[fake_pubkey(62)] = self._derivation_for(
+ wallet_b_fp, 1, 5
+ )
+
+ raw = p.serialize()
+ wallets, meta = self.manager.preprocess_psbt(BytesIO(raw), BytesIO())
+
+ outputs = meta["outputs"]
+ self.assertEqual(len(outputs), 1)
+ self.assertFalse(outputs[0]["change"])
+ self.assertIn("label", outputs[0])
+ self.assertNotIn("change", outputs[0]["label"])
+
+ def test_branch1_output_with_mixed_spending_wallets_is_not_labeled_change(self):
+ # Inputs come from both Wallet A and Wallet B (ambiguous spending
+ # context), and the output verifiably belongs to Wallet A on
+ # branch 1. get_output_status() correctly returns
+ # is_change=False here too (len(wallets) != 1), so again the label
+ # must not say "change".
+ wallet_a = self.wallet
+ wallet_b, wallet_b_fp = self._other_wallet("zoo " * 11 + "wrong")
+
+ txin_a = TransactionInput(b"\x33" * 32, 0)
+ txin_b = TransactionInput(b"\x44" * 32, 0)
+ out0 = TransactionOutput(29_000, self.script_for(wallet_a, 1, 3))
+
+ tx = Transaction(vin=[txin_a, txin_b], vout=[out0])
+ p = PSBT(tx)
+ p.inputs[0].witness_utxo = TransactionOutput(
+ 20_000, self.script_for(wallet_a, 0, 0)
+ )
+ p.inputs[0].bip32_derivations[fake_pubkey(71)] = self.derivation(wallet_a, 0, 0)
+ p.inputs[1].witness_utxo = TransactionOutput(
+ 15_000, wallet_b.descriptor.derive(0, branch_index=0).script_pubkey()
+ )
+ p.inputs[1].bip32_derivations[fake_pubkey(72)] = self._derivation_for(
+ wallet_b_fp, 0, 0
+ )
+ p.outputs[0].bip32_derivations[fake_pubkey(73)] = self.derivation(wallet_a, 1, 3)
+
+ raw = p.serialize()
+ wallets, meta = self.manager.preprocess_psbt(BytesIO(raw), BytesIO())
+
+ self.assertEqual(len(wallets), 2)
+ outputs = meta["outputs"]
+ self.assertEqual(len(outputs), 1)
+ self.assertFalse(outputs[0]["change"])
+ self.assertIn("label", outputs[0])
+ self.assertNotIn("change", outputs[0]["label"])
### test/tests_native/test_change_security.py
@@ -0,0 +1,566 @@
+import ast
+import gc
+import sys
+from io import BytesIO
+from pathlib import Path
+from types import SimpleNamespace
+from unittest import TestCase
+
+if sys.implementation.name != "micropython":
+ from native_support import setup_native_stubs
+
+ setup_native_stubs()
+
+from embit import bip32, ec, script
+from embit.psbt import DerivationPath, PSBT
+from embit.transaction import Transaction, TransactionInput, TransactionOutput
+
+from apps.wallets.wallet import Wallet
+from apps.wallets.manager import UNVERIFIED_CHANGE_WARNING
+from tests.util import clear_testdir, get_keystore, get_wallets_app
+
+
+INVALID_CHANGE_WARNING = (
+ "Invalid change metadata! Host claimed this output as wallet change, "
+ "but it does not match your wallet. Verify the destination."
+)
+
+
+def fake_pubkey(seed):
+ return ec.PrivateKey(bytes([seed]) * 32).get_public_key()
+
+
+class ChangeSecurityTest(TestCase):
+ def setUp(self):
+ clear_testdir()
+ self.keystore = get_keystore()
+ self.app = get_wallets_app(self.keystore, "regtest")
+ self.manager = self.app.manager
+ self.wallet = self.manager.wallets[0]
+ self.fingerprint = self.keystore.fingerprint
+ self.origin = "m/84h/1h/0h"
+
+ def tearDown(self):
+ clear_testdir()
+ gc.collect()
+
+ def derivation(self, branch, index, origin=None):
+ origin = origin or self.origin
+ path = bip32.parse_path("%s/%d/%d" % (origin, branch, index))
+ return DerivationPath(self.fingerprint, path)
+
+ def output(self, branch=None, index=0, output_script=None, taproot=False):
+ derivations = {}
+ taproot_derivations = {}
+ if branch is not None:
+ der = self.derivation(branch, index)
+ if taproot:
+ taproot_derivations[fake_pubkey(1)] = ([], der)
+ else:
+ derivations[fake_pubkey(1)] = der
+ return SimpleNamespace(
+ bip32_derivations=derivations,
+ taproot_bip32_derivations=taproot_derivations,
+ script_pubkey=output_script,
+ )
+
+ def wallet_script(self, wallet, branch, index):
+ return wallet.descriptor.derive(index, branch_index=branch).script_pubkey()
+
+ def test_canonical_change_is_verified_without_warning(self):
+ out = self.output(1, 3, self.wallet_script(self.wallet, 1, 3))
+ status = self.manager.get_output_status(self.wallet, {self.wallet: {}}, out)
+ self.assertEqual(status, ((3, 1), True, None))
+
+ def test_internal_change_address_with_unknown_input_is_not_verified_change(self):
+ out = self.output(1, 10, self.wallet_script(self.wallet, 1, 10))
+ derivation, change, warning = self.manager.get_output_status(
+ self.wallet, {self.wallet: {}, None: {}}, out
+ )
+ self.assertEqual(derivation, (10, 1))
+ self.assertFalse(change)
+ self.assertEqual(
+ warning, UNVERIFIED_CHANGE_WARNING % (1, 10)
+ )
+
+ def test_receive_branch_is_visible_wallet_output_without_warning(self):
+ out = self.output(0, 3, self.wallet_script(self.wallet, 0, 3))
+ derivation, change, warning = self.manager.get_output_status(
+ self.wallet, {self.wallet: {}}, out
+ )
+ self.assertEqual(derivation, (3, 0))
+ self.assertFalse(change)
+ self.assertIsNone(warning)
+
+ def test_external_output_has_no_wallet_metadata_warning(self):
+ out = self.output(output_script=script.p2wpkh(fake_pubkey(44)))
+ self.assertEqual(
+ self.manager.get_output_status(self.wallet, {self.wallet: {}}, out),
+ (None, False, None),
+ )
+
+ def test_forged_change_claim_is_detected_before_ownership_is_discarded(self):
+ attacker_script = script.p2wpkh(fake_pubkey(99))
+ out = self.output(1, 9, attacker_script)
+ # Descriptor.owns() rejects this output, which is why the normal
+ # ownership result is None. The spending-wallet scan still detects
+ # the descriptor-valid branch-1 claim and reports its mismatch.
+ derivation, change, warning = self.manager.get_output_status(
+ None, {self.wallet: {}}, out
+ )
+ self.assertIsNone(derivation)
+ self.assertFalse(change)
+ self.assertEqual(warning, INVALID_CHANGE_WARNING)
+
+ def test_forged_change_claim_is_detected_with_valid_receive_claim(self):
+ receive = self.wallet.descriptor.derive(3, branch_index=0)
+ forged_change = self.wallet.descriptor.derive(9, branch_index=1)
+ out = SimpleNamespace(
+ bip32_derivations={
+ receive.keys[0].get_public_key(): self.derivation(0, 3),
+ forged_change.keys[0].get_public_key(): self.derivation(1, 9),
+ },
+ taproot_bip32_derivations={},
+ script_pubkey=receive.script_pubkey(),
+ )
+ derivation, change, warning = self.manager.get_output_status(
+ self.wallet, {self.wallet: {}}, out
+ )
+ self.assertEqual(derivation, (3, 0))
+ self.assertFalse(change)
+ self.assertEqual(warning, INVALID_CHANGE_WARNING)
+
+ def test_unrelated_wallet_metadata_does_not_warn(self):
+ other_keystore = get_keystore(mnemonic="zoo " * 11 + "wrong")
+ other_origin = "m/84h/1h/0h"
+ other_xpub = other_keystore.get_xpub(other_origin)
+ other_desc = "wpkh([%s%s]%s/<0;1>/*)" % (
+ other_keystore.fingerprint.hex(),
+ other_origin[1:],
+ other_xpub.to_base58(self.manager.Networks["regtest"]["xpub"]),
+ )
+ out = SimpleNamespace(
+ bip32_derivations={
+ fake_pubkey(99): DerivationPath(
+ other_keystore.fingerprint,
+ bip32.parse_path("%s/1/9" % other_origin),
+ )
+ },
+ taproot_bip32_derivations={},
+ script_pubkey=script.p2wpkh(fake_pubkey(99)),
+ )
+ self.assertIsNone(
+ self.manager.get_output_status(None, {self.wallet: {}}, out)[2]
+ )
+
+ def test_conflicting_change_claim_on_other_wallet_output_warns(self):
+ other_keystore = get_keystore(mnemonic="zoo " * 11 + "wrong")
+ other_origin = "m/84h/1h/0h"
+ other_xpub = other_keystore.get_xpub(other_origin)
+ other_desc = "wpkh([%s%s]%s/<0;1>/*)" % (
+ other_keystore.fingerprint.hex(),
+ other_origin[1:],
+ other_xpub.to_base58(self.manager.Networks["regtest"]["xpub"]),
+ )
+ other = Wallet.from_descriptor(other_desc, None)
+ other.name = "Other"
+ self.manager.wallets.append(other)
+
+ other_derived = other.descriptor.derive(3, branch_index=1)
+ forged_change = self.wallet.descriptor.derive(9, branch_index=1)
+ tx = Transaction(
+ vin=[TransactionInput(b"5" * 32, 0)],
+ vout=[TransactionOutput(20_000, other_derived.script_pubkey())],
+ )
+ psbt = PSBT(tx)
+ psbt.inputs[0].witness_utxo = TransactionOutput(
+ 25_000, self.wallet_script(self.wallet, 0, 0)
+ )
+ psbt.inputs[0].bip32_derivations[fake_pubkey(85)] = self.derivation(0, 0)
+ psbt.outputs[0].bip32_derivations[
+ other_derived.keys[0].get_public_key()
+ ] = DerivationPath(
+ other_keystore.fingerprint,
+ bip32.parse_path("%s/1/3" % other_origin),
+ )
+ psbt.outputs[0].bip32_derivations[
+ forged_change.keys[0].get_public_key()
+ ] = self.derivation(1, 9)
+ _, meta = self.manager.preprocess_psbt(BytesIO(psbt.serialize()), BytesIO())
+ output = meta["outputs"][0]
+ self.assertFalse(output["change"])
+ self.assertIn("Other", output["label"])
+ self.assertNotIn("change", output["label"])
+ self.assertIn(INVALID_CHANGE_WARNING, output["warnings"])
+
+ def test_shared_derivation_for_other_wallet_does_not_warn(self):
+ xpub = self.keystore.get_xpub(self.origin)
+ other_desc = "sh(wpkh([%s%s]%s/<0;1>/*))" % (
+ self.fingerprint.hex(),
+ self.origin[1:],
+ xpub.to_base58(self.manager.Networks["regtest"]["xpub"]),
+ )
+ other = Wallet.from_descriptor(other_desc, None)
+ other.name = "Nested"
+ self.manager.wallets.append(other)
+
+ derived = other.descriptor.derive(3, branch_index=1)
+ tx = Transaction(
+ vin=[TransactionInput(b"7" * 32, 0)],
+ vout=[TransactionOutput(20_000, derived.script_pubkey())],
+ )
+ psbt = PSBT(tx)
+ psbt.inputs[0].witness_utxo = TransactionOutput(
+ 25_000, self.wallet_script(self.wallet, 0, 0)
+ )
+ psbt.inputs[0].bip32_derivations[fake_pubkey(88)] = self.derivation(0, 0)
+ psbt.outputs[0].bip32_derivations[
+ derived.keys[0].get_public_key()
+ ] = self.derivation(1, 3)
+
+ _, meta = self.manager.preprocess_psbt(BytesIO(psbt.serialize()), BytesIO())
+ output = meta["outputs"][0]
+ self.assertFalse(output["change"])
+ self.assertIn("Nested", output["label"])
+ self.assertNotIn(INVALID_CHANGE_WARNING, output.get("warnings", []))
+
+ def test_shared_taproot_output_key_derivation_does_not_warn(self):
+ xpub = self.keystore.get_xpub(self.origin)
+ other_desc = "tr([%s%s]%s/<0;1>/*)" % (
+ self.fingerprint.hex(),
+ self.origin[1:],
+ xpub.to_base58(self.manager.Networks["regtest"]["xpub"]),
+ )
+ other = Wallet.from_descriptor(other_desc, None)
+ other.name = "Taproot"
+ self.manager.wallets.append(other)
+
+ derived = other.descriptor.derive(3, branch_index=1)
+ tx = Transaction(
+ vin=[TransactionInput(b"8" * 32, 0)],
+ vout=[TransactionOutput(20_000, derived.script_pubkey())],
+ )
+ psbt = PSBT(tx)
+ psbt.inputs[0].witness_utxo = TransactionOutput(
+ 25_000, self.wallet_script(self.wallet, 0, 0)
+ )
+ psbt.inputs[0].bip32_derivations[fake_pubkey(89)] = self.derivation(0, 0)
+ output_key = ec.PublicKey.from_xonly(derived.script_pubkey().data[2:])
+ psbt.outputs[0].taproot_bip32_derivations[output_key] = (
+ [],
+ self.derivation(1, 3),
+ )
+
+ _, meta = self.manager.preprocess_psbt(BytesIO(psbt.serialize()), BytesIO())
+ output = meta["outputs"][0]
+ self.assertFalse(output["change"])
+ self.assertIn("Taproot", output["label"])
+ self.assertNotIn(INVALID_CHANGE_WARNING, output.get("warnings", []))
+
+ def test_multipath_claim_is_bound_regardless_of_metadata_order(self):
+ origin = "m/48h/1h/0h/2h"
+ xpub = self.keystore.get_xpub(origin)
+ key = "[%s%s]%s" % (
+ self.fingerprint.hex(),
+ origin[1:],
+ xpub.to_base58(self.manager.Networks["regtest"]["xpub"]),
+ )
+ descriptor = "wsh(multi(2,%s/<0;1>/*,%s/<1;0>/*))" % (key, key)
+ wallet = Wallet.from_descriptor(descriptor, None)
+ derived = wallet.descriptor.derive(3, branch_index=0)
+ out = SimpleNamespace(
+ bip32_derivations={
+ derived.keys[0].get_public_key(): DerivationPath(
+ self.fingerprint,
+ bip32.parse_path("%s/0/3" % origin),
+ ),
+ derived.keys[1].get_public_key(): DerivationPath(
+ self.fingerprint,
+ bip32.parse_path("%s/1/3" % origin),
+ ),
+ },
+ taproot_bip32_derivations={},
+ script_pubkey=derived.script_pubkey(),
+ )
+
+ derivation, change, warning = self.manager.get_output_status(
+ wallet, {wallet: {}}, out
+ )
+ self.assertEqual(derivation, (3, 0))
+ self.assertFalse(change)
+ self.assertIsNone(warning)
+
+ out.bip32_derivations = {
+ derived.keys[1].get_public_key(): DerivationPath(
+ self.fingerprint,
+ bip32.parse_path("%s/1/3" % origin),
+ ),
+ derived.keys[0].get_public_key(): DerivationPath(
+ self.fingerprint,
+ bip32.parse_path("%s/0/3" % origin),
+ ),
+ }
+ derivation, change, warning = self.manager.get_output_status(
+ wallet, {wallet: {}}, out
+ )
+ self.assertEqual(derivation, (3, 0))
+ self.assertFalse(change)
+ self.assertIsNone(warning)
+
+ def test_three_branch_descriptors_never_auto_change(self):
+ xpub = self.keystore.get_xpub(self.origin)
+ desc = "wpkh([%s%s]%s/<0;1;2>/*)" % (
+ self.fingerprint.hex(),
+ self.origin[1:],
+ xpub.to_base58(self.manager.Networks["regtest"]["xpub"]),
+ )
+ wallet = Wallet.from_descriptor(desc, None)
+ out = self.output(1, 3, self.wallet_script(wallet, 1, 3))
+ self.assertFalse(
+ self.manager.get_output_status(wallet, {wallet: {}}, out)[1]
+ )
+
+ def test_three_branch_forged_position_one_metadata_does_not_warn(self):
+ # Sole spending wallet has a <0;1;2> descriptor, the host supplies a
+ # descriptor-valid position-1 derivation, but the real output script
+ # is unrelated. The output stays visible and non-change (as always
+ # for a >2-branch descriptor), and because position 1 has no defined
+ # "change" meaning here the device must NOT raise the invalid-change-
+ # metadata warning: the host never claimed change, only a derivation.
+ xpub = self.keystore.get_xpub(self.origin)
+ desc = "wpkh([%s%s]%s/<0;1;2>/*)" % (
+ self.fingerprint.hex(),
+ self.origin[1:],
+ xpub.to_base58(self.manager.Networks["regtest"]["xpub"]),
+ )
+ wallet = Wallet.from_descriptor(desc, None)
+ wallet.name = "Three branch"
+ out = self.output(1, 9, script.p2wpkh(fake_pubkey(99)))
+ derivation, change, warning = self.manager.get_output_status(
+ None, {wallet: {}}, out
+ )
+ self.assertIsNone(derivation)
+ self.assertFalse(change)
+ self.assertIsNone(warning)
+
+ def test_one_branch_descriptor_never_auto_change(self):
+ xpub = self.keystore.get_xpub(self.origin)
+ desc = "wpkh([%s%s]%s/<0>/*)" % (
+ self.fingerprint.hex(),
+ self.origin[1:],
+ xpub.to_base58(self.manager.Networks["regtest"]["xpub"]),
+ )
+ wallet = Wallet.from_descriptor(desc, None)
+ out = self.output(0, 3, self.wallet_script(wallet, 0, 3))
+ self.assertEqual(wallet.descriptor.num_branches, 1)
+ self.assertFalse(self.manager.get_output_status(wallet, {wallet: {}}, out)[1])
+
+ def test_unusual_three_branch_position_one_never_auto_change(self):
+ xpub = self.keystore.get_xpub(self.origin)
+ desc = "wpkh([%s%s]%s/<22;33;44>/*)" % (
+ self.fingerprint.hex(),
+ self.origin[1:],
+ xpub.to_base58(self.manager.Networks["regtest"]["xpub"]),
+ )
+ wallet = Wallet.from_descriptor(desc, None)
+ der = DerivationPath(
+ self.fingerprint,
+ bip32.parse_path("%s/33/4" % self.origin),
+ )
+ out = SimpleNamespace(
+ bip32_derivations={fake_pubkey(2): der},
+ taproot_bip32_derivations={},
+ script_pubkey=self.wallet_script(wallet, 1, 4),
+ )
+ self.assertEqual(wallet.descriptor.num_branches, 3)
+ self.assertFalse(self.manager.get_output_status(wallet, {wallet: {}}, out)[1])
+
+ def test_two_branch_unusual_raw_value_uses_position_one(self):
+ xpub = self.keystore.get_xpub(self.origin)
+ desc = "wpkh([%s%s]%s/<22;33>/*)" % (
+ self.fingerprint.hex(),
+ self.origin[1:],
+ xpub.to_base58(self.manager.Networks["regtest"]["xpub"]),
+ )
+ wallet = Wallet.from_descriptor(desc, None)
+ der = DerivationPath(
+ self.fingerprint,
+ bip32.parse_path("%s/33/4" % self.origin),
+ )
+ out = SimpleNamespace(
+ bip32_derivations={fake_pubkey(3): der},
+ taproot_bip32_derivations={},
+ script_pubkey=self.wallet_script(wallet, 1, 4),
+ )
+ self.assertTrue(self.manager.get_output_status(wallet, {wallet: {}}, out)[1])
+
+ def test_taproot_change_and_forged_taproot_claim(self):
+ origin = "m/86h/1h/0h"
+ xpub = self.keystore.get_xpub(origin)
+ desc = "tr([%s%s]%s/<0;1>/*)" % (
+ self.fingerprint.hex(),
+ origin[1:],
+ xpub.to_base58(self.manager.Networks["regtest"]["xpub"]),
+ )
+ wallet = Wallet.from_descriptor(desc, None)
+ der = DerivationPath(self.fingerprint, bip32.parse_path("%s/1/3" % origin))
+ out = SimpleNamespace(
+ bip32_derivations={},
+ taproot_bip32_derivations={fake_pubkey(4): ([], der)},
+ script_pubkey=self.wallet_script(wallet, 1, 3),
+ )
+ self.assertTrue(self.manager.get_output_status(wallet, {wallet: {}}, out)[1])
+ out.script_pubkey = script.p2wpkh(fake_pubkey(88))
+ self.assertEqual(
+ self.manager.get_output_status(None, {wallet: {}}, out)[2],
+ INVALID_CHANGE_WARNING,
+ )
+
+ def test_multiple_spending_wallets_and_unknown_wallet_fail_closed(self):
+ other = Wallet.from_descriptor(str(self.wallet.descriptor), None)
+ out = self.output(1, 3, self.wallet_script(self.wallet, 1, 3))
+ self.assertFalse(
+ self.manager.get_output_status(
+ self.wallet, {self.wallet: {}, other: {}}, out
+ )[1]
+ )
+ self.assertFalse(
+ self.manager.get_output_status(self.wallet, {None: {}}, out)[1]
+ )
+
+ def test_output_warnings_are_accumulated(self):
+ metaout = {"warnings": ["Existing warning!"]}
+ self.manager.add_output_warning(metaout, INVALID_CHANGE_WARNING)
+ self.manager.add_output_warning(metaout, "Existing warning!")
+ self.assertEqual(
+ metaout["warnings"], ["Existing warning!", INVALID_CHANGE_WARNING]
+ )
+ self.assertNotIn("warning", metaout)
+
+ def test_exact_manual_five_transaction_warns_through_preprocess(self):
+ wallet = self.wallet
+ tx = Transaction(
+ vin=[TransactionInput(b"3" * 32, 0)],
+ vout=[
+ TransactionOutput(95_000, script.p2wpkh(fake_pubkey(70))),
+ TransactionOutput(24_000, script.p2wpkh(fake_pubkey(71))),
+ ],
+ )
+ psbt = PSBT(tx)
+ psbt.inputs[0].witness_utxo = TransactionOutput(
+ 120_000, self.wallet_script(wallet, 0, 0)
+ )
+ psbt.inputs[0].bip32_derivations[fake_pubkey(72)] = self.derivation(0, 0)
+ # The metadata claims /1/9, while output 1 is the attacker script.
+ psbt.outputs[1].bip32_derivations[fake_pubkey(73)] = self.derivation(1, 9)
+ _, meta = self.manager.preprocess_psbt(BytesIO(psbt.serialize()), BytesIO())
+ output = meta["outputs"][1]
+ self.assertFalse(output["change"])
+ self.assertNotIn("label", output)
+ self.assertEqual(output["value"], 24_000)
+ self.assertEqual(output["address"], self.manager.get_address(tx.vout[1]))
+ self.assertEqual(output["warnings"], [INVALID_CHANGE_WARNING])
+
+ def test_unknown_input_internal_change_warning_survives_preprocess(self):
+ wallet = self.wallet
+ tx = Transaction(
+ vin=[TransactionInput(b"6" * 32, 0)],
+ vout=[
+ TransactionOutput(
+ 95_000, self.wallet_script(wallet, 1, 10)
+ )
+ ],
+ )
+ psbt = PSBT(tx)
+ psbt.inputs[0].witness_utxo = TransactionOutput(
+ 120_000, script.p2wpkh(fake_pubkey(91))
+ )
+ psbt.outputs[0].bip32_derivations[fake_pubkey(92)] = self.derivation(1, 10)
+ _, meta = self.manager.preprocess_psbt(BytesIO(psbt.serialize()), BytesIO())
+ output = meta["outputs"][0]
+ self.assertFalse(output["change"])
+ self.assertEqual(
+ output["warnings"], [UNVERIFIED_CHANGE_WARNING % (1, 10)]
+ )
+
+ def test_two_suspicious_outputs_keep_warnings_separate(self):
+ wallet = self.wallet
+ tx = Transaction(
+ vin=[TransactionInput(b"4" * 32, 0)],
+ vout=[
+ TransactionOutput(10_000, script.p2wpkh(fake_pubkey(80))),
+ TransactionOutput(20_000, script.p2wpkh(fake_pubkey(81))),
+ ],
+ )
+ psbt = PSBT(tx)
+ psbt.inputs[0].witness_utxo = TransactionOutput(
+ 40_000, self.wallet_script(wallet, 0, 0)
+ )
+ psbt.inputs[0].bip32_derivations[fake_pubkey(82)] = self.derivation(0, 0)
+ psbt.outputs[0].bip32_derivations[fake_pubkey(83)] = self.derivation(1, 4)
+ psbt.outputs[1].bip32_derivations[fake_pubkey(84)] = self.derivation(1, 5)
+ _, meta = self.manager.preprocess_psbt(BytesIO(psbt.serialize()), BytesIO())
+ self.assertEqual(
+ [out["warnings"] for out in meta["outputs"]],
+ [[INVALID_CHANGE_WARNING], [INVALID_CHANGE_WARNING]],
+ )
+
+ def test_liquid_uses_the_same_conservative_change_rules(self):
+ clear_testdir()
+ liquid_keystore = get_keystore()
+ liquid_app = get_wallets_app(liquid_keystore, "elementsregtest")
+ liquid_manager = liquid_app.manager
+ liquid_wallet = liquid_manager.wallets[0]
+ origin = "m/84h/1h/0h"
+
+ def liquid_output(branch, index):
+ der = DerivationPath(
+ liquid_keystore.fingerprint,
+ bip32.parse_path("%s/%d/%d" % (origin, branch, index)),
+ )
+ return SimpleNamespace(
+ bip32_derivations={fake_pubkey(90 + branch): der},
+ taproot_bip32_derivations={},
+ script_pubkey=liquid_wallet.descriptor.derive(
+ index, branch_index=branch
+ ).script_pubkey(),
+ )
+
+ self.assertTrue(
+ liquid_manager.get_output_status(
+ liquid_wallet, {liquid_wallet: {}}, liquid_output(1, 3)
+ )[1]
+ )
+ self.assertFalse(
+ liquid_manager.get_output_status(
+ liquid_wallet, {liquid_wallet: {}}, liquid_output(0, 3)
+ )[1]
+ )
+
+
+class TransactionScreenSecurityTest(TestCase):
+ """Guard the primary/details visibility contract without an LVGL device."""
+
+ def setUp(self):
+ path = Path(__file__).resolve().parents[2] / "src" / "gui" / "screens" / "transaction.py"
+ self.tree = ast.parse(path.read_text())
+
+ def test_primary_page_shows_warning_bearing_outputs(self):
+ guards = [
+ node
+ for node in ast.walk(self.tree)
+ if isinstance(node, ast.If)
+ and ast.unparse(node.test) == "out['change'] and (not out.get('warnings'))"
+ ]
+ self.assertEqual(len(guards), 1)
+ self.assertTrue(any(isinstance(node, ast.Continue) for node in guards[0].body))
+ source = ast.unparse(self.tree)
+ self.assertIn("self.show_output(out, obj)", source)
+ self.assertIn("warning_text = '\\n'.join(out.get('warnings', []))", source)
+
+ def test_details_page_has_no_output_skip(self):
+ for node in ast.walk(self.tree):
+ if isinstance(node, ast.For) and isinstance(node.target, ast.Tuple):
+ if any(isinstance(part, ast.Name) and part.id == "out" for part in node.target.elts):
+ for child in ast.walk(node):
+ self.assertNotIsInstance(child, (ast.Continue, ast.Break))
### test/tests_native/test_transaction_confirmation.py
@@ -0,0 +1,190 @@
+import ast
+from pathlib import Path
+from unittest import TestCase
+
+# gui.screens.transaction.TransactionScreen builds real LVGL widgets (pages,
+# labels, switches, styles) directly in __init__, and the native test stubs
+# in native_support.py replace gui.screens.TransactionScreen with a trivial
+# placeholder class (as they do for the other screen classes) so the rest of
+# the wallet-manager code can be exercised without a display. That means the
+# real screen implementation cannot be instantiated - or its widget tree
+# inspected - under the current native/unix test infrastructure.
+#
+# The UX/security invariant this file guards, after reviewing PR #13's
+# original "always show every output" behaviour against how BitBox02
+# handles the same distinction:
+#
+# - An output that WalletManager.get_output_status() has
+# cryptographically verified as change (descriptor branch 1, and an
+# on-device re-derivation of the script_pubkey that matches the actual
+# output) and that carries no warning is *not* shown individually on
+# the primary confirmation page. The device itself already proved this
+# output can't be an attacker-controlled destination, so asking the
+# user to re-check it adds noise without adding security. It stays
+# fully visible on the details page ("Show detailed information"),
+# which lists every output unconditionally.
+# - Every other output is always shown on the primary page: external or
+# unverifiable recipients (the actual security-relevant case - a
+# malicious host must not be able to hide an injected output), and
+# same-wallet outputs that are *not* on the verified change branch
+# (e.g. a receive-branch self-payment, which WalletManager labels
+# "This wallet (...)" rather than leaving it looking like automatic
+# change). A "change" output that carries a warning (e.g. gap-limit
+# exceeded) is shown too, since the warning means it needs attention.
+#
+# Since we cannot drive the real widget tree here, this test statically
+# proves (via the AST, not a text/regex match) that the loop over
+# meta["outputs"] in the primary confirmation section of
+# TransactionScreen.__init__ skips an output if and only if it is guarded
+# by exactly `out["change"] and not out.get("warnings")`, with no other
+# conditional skip anywhere in the loop.
+
+TRANSACTION_SCREEN_PATH = (
+ Path(__file__).resolve().parents[2] / "src" / "gui" / "screens" / "transaction.py"
+)
+
+
+class TransactionConfirmationVisibilityTest(TestCase):
+ def setUp(self):
+ self.source = TRANSACTION_SCREEN_PATH.read_text()
+ self.tree = ast.parse(self.source)
+
+ def _find_class(self, name):
+ for node in ast.walk(self.tree):
+ if isinstance(node, ast.ClassDef) and node.name == name:
+ return node
+ self.fail("class %s not found in %s" % (name, TRANSACTION_SCREEN_PATH))
+
+ def _find_init(self, class_node):
+ for node in class_node.body:
+ if isinstance(node, ast.FunctionDef) and node.name == "__init__":
+ return node
+ self.fail("__init__ not found on TransactionScreen")
+
+ def _primary_output_loop(self, init_node):
+ # the primary-confirmation loop is `for out in meta["outputs"]: ...`
+ for node in ast.walk(init_node):
+ if isinstance(node, ast.For) and isinstance(node.target, ast.Name) and node.target.id == "out":
+ if (
+ isinstance(node.iter, ast.Subscript)
+ and isinstance(node.iter.value, ast.Name)
+ and node.iter.value.id == "meta"
+ ):
+ return node
+ self.fail('for out in meta["outputs"]: loop not found in TransactionScreen.__init__')
+
+ def _is_out_subscript_of(self, node, name, key):
+ return (
+ isinstance(node, ast.Subscript)
+ and isinstance(node.value, ast.Name)
+ and node.value.id == name
+ and isinstance(node.slice, (ast.Constant, ast.Str))
+ and getattr(node.slice, "value", getattr(node.slice, "s", None)) == key
+ )
+
+ def _is_change_guard(self, test):
+ # out["change"] and not out.get("warnings")
+ if not (isinstance(test, ast.BoolOp) and isinstance(test.op, ast.And)):
+ return False
+ if len(test.values) != 2:
+ return False
+ change_check, warning_check = test.values
+ if not self._is_out_subscript_of(change_check, "out", "change"):
+ return False
+ if not (isinstance(warning_check, ast.UnaryOp) and isinstance(warning_check.op, ast.Not)):
+ return False
+ call = warning_check.operand
+ return (
+ isinstance(call, ast.Call)
+ and isinstance(call.func, ast.Attribute)
+ and call.func.attr == "get"
+ and isinstance(call.func.value, ast.Name)
+ and call.func.value.id == "out"
+ and call.args
+ and self._is_str_arg(call.args[0], "warnings")
+ )
+
+ def _is_str_arg(self, node, value):
+ return (
+ isinstance(node, (ast.Constant, ast.Str))
+ and getattr(node, "value", getattr(node, "s", None)) == value
+ )
+
+ def test_primary_confirmation_loop_only_skips_verified_change_without_warning(self):
+ cls = self._find_class("TransactionScreen")
+ init = self._find_init(cls)
+ loop = self._primary_output_loop(init)
+
+ guard_stmts = [stmt for stmt in loop.body if isinstance(stmt, ast.If)]
+ self.assertEqual(
+ len(guard_stmts), 1,
+ "primary confirmation loop must have exactly one skip guard",
+ )
+ guard = guard_stmts[0]
+ self.assertTrue(
+ self._is_change_guard(guard.test),
+ 'the only skip guard must be exactly `out["change"] and not out.get("warnings")` '
+ '- a verified change output without a warning, and nothing else, may be hidden',
+ )
+ self.assertTrue(
+ any(isinstance(s, ast.Continue) for s in guard.body),
+ "the change guard must skip past show_output via continue",
+ )
+ self.assertEqual(
+ len(guard.orelse), 0,
+ "the change guard must not have an else branch hiding other logic",
+ )
+
+ # every other statement in the loop body must call
+ # self.show_output(out, obj) unconditionally - external, unverified,
+ # and same-wallet-but-not-change outputs are never skipped.
+ other_stmts = [stmt for stmt in loop.body if stmt is not guard]
+ for stmt in other_stmts:
+ self.assertNotIsInstance(
+ stmt,
+ (ast.If, ast.Continue, ast.Break),
+ "no output other than verified, warning-free change may be conditionally skipped",
+ )
+ calls_show_output = any(
+ isinstance(stmt, ast.Assign)
+ and isinstance(stmt.value, ast.Call)
+ and isinstance(stmt.value.func, ast.Attribute)
+ and stmt.value.func.attr == "show_output"
+ for stmt in other_stmts
+ )
+ self.assertTrue(
+ calls_show_output,
+ "primary confirmation loop must call self.show_output() for every non-change (or warned) output",
+ )
+
+ def test_details_page_still_lists_every_output_unconditionally(self):
+ # the details page ("Show detailed information") is the fallback
+ # that keeps a hidden verified-change output inspectable; it must
+ # keep iterating meta["outputs"] without a change-keyed skip.
+ cls = self._find_class("TransactionScreen")
+ init = self._find_init(cls)
+
+ details_loops = [
+ node
+ for node in ast.walk(init)
+ if isinstance(node, ast.For)
+ and isinstance(node.iter, ast.Call)
+ and isinstance(node.iter.func, ast.Name)
+ and node.iter.func.id == "enumerate"
+ and node.iter.args
+ and self._is_out_subscript_of(node.iter.args[0], "meta", "outputs")
+ ]
+ self.assertEqual(
+ len(details_loops), 1,
+ 'for i, out in enumerate(meta["outputs"]): loop not found on the details page',
+ )
+ loop = details_loops[0]
+ # cosmetic `if out.get("label", ""): ... else: ...` / warning-text
+ # branches (styling, optional warning label) are fine here
+ # - only a continue/break would actually skip an output.
+ for node in ast.walk(loop):
+ self.assertNotIsInstance(
+ node,
+ (ast.Continue, ast.Break),
+ "the details page must list every output, including hidden verified change",
+ )Why this scored 78/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.