Merge pull request #662 from Foundation-Devices/fix/unverified-psbt-fees

### CHANGELOG.md
@@ -5,6 +5,7 @@ SPDX-License-Identifier: GPL-3.0-or-later
 -->
 
 ## Head
+- Identify network fees from unverifiable PSBT inputs as unverified
 - Validate the complete local xpub when importing multisig wallets
 - Require confirmation before using PSBT-proposed multisig wallets with temporary seeds,
   and cancel signing if import is declined

### ports/stm32/boards/Passport/modules/flows/sign_psbt_common_flow.py
@@ -260,7 +260,9 @@ def render_warnings(self):
             # gc.collect()
 
             fee = self.psbt.calculate_fee()
-            if fee is not None:
+            if not self.psbt.fee_is_verified:
+                msg.write('\n{}\nUnverified '.format(recolor(HIGHLIGHT_TEXT_HEX, 'Network Fee')))
+            elif fee is not None:
                 amount, units = self.chain.render_value(fee)
                 msg.write('\n{}\n{} {} '.format(recolor(HIGHLIGHT_TEXT_HEX, 'Network Fee'), amount, units))
 

### ports/stm32/boards/Passport/modules/psbt.py
@@ -643,19 +643,19 @@ def get_utxo(self, idx):
         fd = self.fd
         old_pos = fd.tell()
 
+        witness_utxo = None
         if self.witness_utxo:
-            # Going forward? Just what we will witness; no other junk
-            # - prefer this format, altho does that imply segwit txn must be generated?
-            # - I don't know why we wouldn't always use this
-            # - once we use this partial utxo data, we must create witness data out
+            # Load the compact output. If the full previous transaction is also
+            # present, its hash-bound output is loaded below and must match.
             self.is_segwit = True
 
             fd.seek(self.witness_utxo[0])
-            utxo = CTxOut()
-            utxo.deserialize(fd)
-            fd.seek(old_pos)
+            witness_utxo = CTxOut()
+            witness_utxo.deserialize(fd)
 
-            return utxo
+            if not self.utxo:
+                fd.seek(old_pos)
+                return witness_utxo
 
         assert self.utxo, 'no utxo'
 
@@ -685,6 +685,11 @@ def get_utxo(self, idx):
 
         fd.seek(old_pos)
 
+        if witness_utxo:
+            assert witness_utxo.nValue == utxo.nValue and \
+                witness_utxo.scriptPubKey == utxo.scriptPubKey, \
+                "witness/non-witness UTXO mismatch for input #%d" % idx
+
         return utxo
 
     def determine_my_signing_key(self, my_idx, utxo, my_xfp, psbt):
@@ -852,6 +857,40 @@ def determine_my_signing_key(self, my_idx, utxo, my_xfp, psbt):
         # Could probably free self.subpaths and self.redeem_script now, but only if we didn't
         # need to re-serialize as a PSBT.
 
+    def get_signing_node(self, sv, my_xfp, my_idx):
+        if self.is_multisig:
+            # The fingerprint is only a hint. Derive each candidate to prove
+            # that Passport owns one of the public keys required by the script.
+            for which_key in self.required_key:
+                skp = keypath_to_str(self.subpaths[which_key])
+                node = sv.derive_path(skp, register=True)
+                if node.public_key() == which_key:
+                    return node, which_key
+
+            raise AssertionError("Input #%d needs pubkey this Passport doesn't have." % my_idx)
+
+        which_key = self.required_key
+
+        if self.subpaths and \
+                (self.subpaths[which_key][0] == my_xfp or
+                 self.subpaths[which_key][0] == swab32(my_xfp)):
+            skp = keypath_to_str(self.subpaths[which_key])
+            node = sv.derive_path(skp, register=True)
+            pubkey = node.public_key()
+        elif self.tap_subpaths and \
+                (self.tap_subpaths[which_key][0][0] == my_xfp or
+                 self.tap_subpaths[which_key][0][0] == swab32(my_xfp)):
+            skp = keypath_to_str(self.tap_subpaths[which_key][0])
+            node = sv.derive_path(skp, register=True)
+            pubkey = node.public_key()[1:]
+        else:
+            raise AssertionError("Input #%d has no Passport signing path." % my_idx)
+
+        if pubkey != which_key:
+            raise AssertionError("Path (%s) led to wrong pubkey for input #%d" % (skp, my_idx))
+
+        return node, which_key
+
     def store(self, kt, key, val):
         # Capture what we are interested in.
 
@@ -957,6 +996,7 @@ def __init__(self):
         self.lock_time = None
         self.total_value_out = None
         self.total_value_in = None
+        self.fee_is_verified = True
         self.presigned_inputs = set()
         self.multisig_import_needs_approval = False
         self.self_send = False
@@ -1269,15 +1309,21 @@ def consider_outputs(self):
         # print('total_non_change_out={} self.total_value_out={}  total_change={}'.format(total_non_change_out,
         #       self.total_value_out, total_change))
         fee = self.calculate_fee()
+        per_fee = None
         if self.total_value_out == 0:
             per_fee = 100
         elif total_non_change_out == 0:
             self.self_send = True
-        else:
+        elif self.fee_is_verified:
             # Calculate fee based on non-change output value
             per_fee = (fee / total_non_change_out) * 100
 
-        if self.self_send:
+        if not self.fee_is_verified:
+            self.warnings.append(
+                ('Unverified Fee',
+                 'One or more input amounts were supplied by another wallet, '
+                 'so Passport cannot verify the network fee.'))
+        elif self.self_send:
             # self.warnings.append(('Self-Send', 'All outputs are being sent back to this wallet.'))
             per_fee = (fee / self.total_value_out) * 100
             if per_fee >= 5:
@@ -1399,6 +1445,7 @@ def consider_inputs(self):
         # Important: parse incoming UTXO to build total input value
         missing = 0
         total_in = 0
+        witness_inputs_to_verify = []
 
         for i, txi in self.input_iter():
             gc.collect()
@@ -1427,16 +1474,34 @@ def consider_inputs(self):
             # - also finds appropriate multisig wallet to be used
             inp.determine_my_signing_key(i, utxo, self.my_xfp, self)
 
+            if inp.witness_utxo and not inp.utxo:
+                # A false amount for an input we prove we sign makes our
+                # signature invalid, and the history cache catches changed
+                # amounts across attempts. Derivation proves the PSBT's
+                # ownership metadata before that argument is applied.
+                if inp.num_our_keys and inp.required_key:
+                    witness_inputs_to_verify.append(i)
+                else:
+                    self.fee_is_verified = False
+
             gc.collect()
 
-            # iff to UTXO is segwit, then check it's value, and also
-            # capture that value, since it's supposed to be immutable
+            del utxo
+
+        if witness_inputs_to_verify:
+            import stash
+            with stash.SensitiveValues() as sv:
+                for i in witness_inputs_to_verify:
+                    self.inputs[i].get_signing_node(sv, self.my_xfp, i)
+
+        # Only update the amount cache after all claimed owned inputs have
+        # proved their signing keys, so rejected metadata cannot be recorded.
+        for i, txi in self.input_iter():
+            inp = self.inputs[i]
             if inp.is_segwit:
                 history.verify_amount(txi.prevout, inp.amount, i)
                 gc.collect()
 
-            del utxo
-
         gc.collect()
 
         # XXX scan witness data provided, and consider those ins signed if not multisig?

### ports/stm32/boards/Passport/modules/tasks/sign_psbt_task.py
@@ -13,7 +13,6 @@
 async def sign_psbt_task(on_done, psbt):
     from exceptions import FraudulentChangeOutput, FatalPSBTIssue
     from errors import Error
-    from utils import keypath_to_str, swab32
     from serializations import ser_sig_der
     import stash
     import gc
@@ -47,6 +46,10 @@ async def sign_psbt_task(on_done, psbt):
                     # but in other cases, no more signatures are possible
                     continue
 
+                if not inp.is_multisig:
+                    assert not (inp.added_sig or inp.tap_key_sig), \
+                        "This transaction has already been signed"
+
                 txi.scriptSig = inp.scriptSig
                 if not txi.scriptSig:
                     raise AssertionError('No scriptsig?')
@@ -62,51 +65,7 @@ async def sign_psbt_task(on_done, psbt):
                     digest = psbt.make_txn_segwit_sighash(in_idx, txi,
                                                           inp.amount, inp.scriptCode, inp.sighash)
 
-                if inp.is_multisig:
-                    # need to consider a set of possible keys, since xfp may not be unique
-                    for which_key in inp.required_key:
-                        # get node required
-                        skp = keypath_to_str(inp.subpaths[which_key])
-                        node = sv.derive_path(skp, register=False)
-
-                        # expensive test, but works... and important
-                        pu = node.public_key()
-                        if pu == which_key:
-                            break
-                    else:
-                        raise AssertionError("Input #%d needs pubkey this Passport doesn't have." % in_idx)
-
-                else:
-                    # single pubkey <=> single key
-                    which_key = inp.required_key
-
-                    assert not (inp.added_sig or inp.tap_key_sig), "This transaction has already been signed"
-
-                    if len(inp.subpaths) > 0 and \
-                        (inp.subpaths[which_key][0] == psbt.my_xfp or
-                         inp.subpaths[which_key][0] == swab32(psbt.my_xfp)):
-
-                        # get node required
-                        skp = keypath_to_str(inp.subpaths[which_key])
-                        node = sv.derive_path(skp, register=False)
-
-                        # expensive test, but works... and important
-                        pu = node.public_key()
-
-                    # tap_subpaths have type ([path_elements], [tap_hashes])
-                    elif len(inp.tap_subpaths) > 0 and \
-                        (inp.tap_subpaths[which_key][0][0] == psbt.my_xfp or
-                         inp.tap_subpaths[which_key][0][0] == swab32(psbt.my_xfp)):
-
-                        # get node required
-                        skp = keypath_to_str(inp.tap_subpaths[which_key][0])
-                        node = sv.derive_path(skp, register=False)
-
-                        # expensive test, but works... and important
-                        pu = node.public_key()[1:]
-
-                    if pu != which_key:
-                        raise AssertionError("Path (%s) led to wrong pubkey for input #%d" % (skp, in_idx))
+                node, which_key = inp.get_signing_node(sv, psbt.my_xfp, in_idx)
 
                 # The precious private key we need
                 pk = node.private_key()
@@ -137,7 +96,7 @@ async def sign_psbt_task(on_done, psbt):
                 # private key no longer required
                 stash.blank_object(pk)
                 stash.blank_object(node)
-                del pk, node, pu, skp
+                del pk, node
 
                 # print("result %s" % b2a_hex(result).decode('ascii'))
 

### ports/stm32/boards/Passport/modules/tests/test_unit.py
@@ -42,3 +42,7 @@ def test_psbt_amounts(test):
 
 def test_multisig_xpub_validation(test):
     assert test('multisig_xpub_validation.py') == b'OK'
+
+
+def test_psbt_fee(test):
+    assert test('psbt_fee.py') == b'OK'

### ports/stm32/boards/Passport/modules/tests/unit/psbt_amounts.py
@@ -37,6 +37,7 @@ def __init__(self, value):
         self.num_our_keys = 1
         self.required_key = b'key'
         self.is_segwit = False
+        self.witness_utxo = False
 
     def has_utxo(self):
         return True

### ports/stm32/boards/Passport/modules/tests/unit/psbt_fee.py
@@ -0,0 +1,242 @@
+# SPDX-FileCopyrightText: © 2026 Foundation Devices, Inc. <hello@foundation.xyz>
+# SPDX-License-Identifier: GPL-3.0-or-later
+#
+# Test PSBT input-value validation and fee review rendering.
+
+from uio import BytesIO
+from ustruct import pack
+
+import history
+import stash
+from flows.sign_psbt_common_flow import SignPsbtCommonFlow
+from psbt import psbtInputProxy, psbtObject
+from public_constants import (
+    PSBT_IN_BIP32_DERIVATION,
+    PSBT_IN_NON_WITNESS_UTXO,
+    PSBT_IN_WITNESS_UTXO,
+)
+from serializations import CTxOut, hash160, ser_compact_size
+
+
+P2WPKH_SCRIPT = b'\x00\x14' + (b'\x11' * 20)
+MY_XFP = 0x12345678
+OWNED_PUBKEY = b'\x02' + (b'\x55' * 32)
+FORGED_PUBKEY = b'\x03' + (b'\x66' * 32)
+TAPROOT_PUBKEY = b'\x77' * 32
+DERIVED_PUBKEYS = {
+    'm/0': OWNED_PUBKEY,
+    'm/1': b'\x02' + TAPROOT_PUBKEY,
+}
+
+
+def psbt_field(key_type, value, key=b''):
+    full_key = bytes([key_type]) + key
+    return ser_compact_size(len(full_key)) + full_key + ser_compact_size(len(value)) + value
+
+
+def previous_tx(txout):
+    txin = (b'\x00' * 32) + pack('<I', 0) + b'\x00' + pack('<I', 0xffffffff)
+    return pack('<i', 2) + b'\x01' + txin + b'\x01' + txout.serialize() + pack('<I', 0)
+
+
+def make_input(witness_txout, non_witness_txout=None):
+    data = b''
+    if non_witness_txout:
+        data += psbt_field(PSBT_IN_NON_WITNESS_UTXO, previous_tx(non_witness_txout))
+    data += psbt_field(PSBT_IN_WITNESS_UTXO, witness_txout.serialize())
+    return psbtInputProxy(BytesIO(data + b'\x00'), 0)
+
+
+def make_owned_input(pubkey=OWNED_PUBKEY):
+    txout = CTxOut(2000, b'\x00\x14' + hash160(pubkey))
+    data = psbt_field(PSBT_IN_WITNESS_UTXO, txout.serialize())
+    data += psbt_field(PSBT_IN_BIP32_DERIVATION, pack('<II', MY_XFP, 0), pubkey)
+    return psbtInputProxy(BytesIO(data + b'\x00'), 0)
+
+
+def assert_raises(exc_type, callback):
+    try:
+        callback()
+    except exc_type:
+        return
+    raise AssertionError('Expected {}'.format(exc_type))
+
+
+matching = CTxOut(1000, P2WPKH_SCRIPT)
+matching_input = make_input(matching, matching)
+loaded = matching_input.get_utxo(0)
+assert loaded.nValue == matching.nValue
+assert loaded.scriptPubKey == matching.scriptPubKey
+assert matching_input.is_segwit
+
+amount_mismatch = make_input(CTxOut(999, P2WPKH_SCRIPT), matching)
+assert_raises(AssertionError, lambda: amount_mismatch.get_utxo(0))
+
+script_mismatch = make_input(CTxOut(1000, b'\x00\x14' + (b'\x33' * 20)), matching)
+assert_raises(AssertionError, lambda: script_mismatch.get_utxo(0))
+
+
+class FakePrevout:
+    n = 0
+
+
+class FakeTxIn:
+    prevout = FakePrevout()
+
+
+class FakeNode:
+    def __init__(self, public_key):
+        self._public_key = public_key
+
+    def public_key(self):
+        return self._public_key
+
+
+class FakeSensitiveValues:
+    def __enter__(self):
+        return self
+
+    def __exit__(self, _exc_type, _exc, _traceback):
+        pass
+
+    @staticmethod
+    def derive_path(path, register=True):
+        assert register
+        return FakeNode(DERIVED_PUBKEYS[path])
+
+
+class FakeSigningInput:
+    pass
+
+
+class FakeInputPSBT:
+    def __init__(self, psbt_input, my_xfp=0):
+        self.inputs = [psbt_input]
+        self.my_xfp = my_xfp
+        self.total_value_in = None
+        self.fee_is_verified = True
+        self.presigned_inputs = set()
+        self.num_inputs = 1
+        self.warnings = []
+
+    def input_iter(self):
+        yield 0, FakeTxIn()
+
+
+verified_amounts = []
+original_verify_amount = history.verify_amount
+original_sensitive_values = stash.SensitiveValues
+history.verify_amount = lambda _prevout, amount, idx: verified_amounts.append((amount, idx))
+stash.SensitiveValues = FakeSensitiveValues
+try:
+    external_input_psbt = FakeInputPSBT(make_input(CTxOut(2000, P2WPKH_SCRIPT)))
+    psbtObject.consider_inputs(external_input_psbt)
+    assert not external_input_psbt.fee_is_verified
+
+    owned_input = make_owned_input()
+    owned_input.validate(0, FakeTxIn(), MY_XFP)
+    owned_input_psbt = FakeInputPSBT(owned_input, MY_XFP)
+    psbtObject.consider_inputs(owned_input_psbt)
+    assert owned_input_psbt.fee_is_verified
+    assert owned_input.num_our_keys == 1
+    assert owned_input.required_key == OWNED_PUBKEY
+
+    forged_input = make_owned_input(FORGED_PUBKEY)
+    forged_input.validate(0, FakeTxIn(), MY_XFP)
+    forged_input_psbt = FakeInputPSBT(forged_input, MY_XFP)
+    assert_raises(AssertionError, lambda: psbtObject.consider_inputs(forged_input_psbt))
+finally:
+    history.verify_amount = original_verify_amount
+    stash.SensitiveValues = original_sensitive_values
+
+assert verified_amounts == [(2000, 0), (2000, 0)]
+
+multisig_input = FakeSigningInput()
+multisig_input.is_multisig = True
+multisig_input.required_key = {OWNED_PUBKEY}
+multisig_input.subpaths = {OWNED_PUBKEY: [MY_XFP, 0]}
+node, which_key = psbtInputProxy.get_signing_node(
+    multisig_input, FakeSensitiveValues(), MY_XFP, 0)
+assert node.public_key() == OWNED_PUBKEY
+assert which_key == OWNED_PUBKEY
+
+taproot_input = FakeSigningInput()
+taproot_input.is_multisig = False
+taproot_input.required_key = TAPROOT_PUBKEY
+taproot_input.subpaths = {}
+taproot_input.tap_subpaths = {TAPROOT_PUBKEY: ([MY_XFP, 1], [])}
+node, which_key = psbtInputProxy.get_signing_node(
+    taproot_input, FakeSensitiveValues(), MY_XFP, 0)
+assert node.public_key()[1:] == TAPROOT_PUBKEY
+assert which_key == TAPROOT_PUBKEY
+
+missing_path_input = FakeSigningInput()
+missing_path_input.is_multisig = False
+missing_path_input.required_key = OWNED_PUBKEY
+missing_path_input.subpaths = {}
+missing_path_input.tap_subpaths = {}
+assert_raises(
+    AssertionError,
+    lambda: psbtInputProxy.get_signing_node(
+        missing_path_input, FakeSensitiveValues(), MY_XFP, 0),
+)
+
+
+class FakeOutputProxy:
+    is_change = False
+
+    def validate(self, _idx, _txout, _xfp, _active_multisig):
+        pass
+
+
+class FakeOutputPSBT:
+    def __init__(self, fee_is_verified):
+        self.outputs = [FakeOutputProxy()]
+        self.total_value_out = 1000
+        self.total_value_in = 5000
+        self.fee_is_verified = fee_is_verified
+        self.self_send = False
+        self.warnings = []
+        self.my_xfp = 0
+        self.active_multisig = None
+
+    def output_iter(self):
+        yield 0, CTxOut(self.total_value_out, P2WPKH_SCRIPT)
+
+    def calculate_fee(self):
+        return self.total_value_in - self.total_value_out
+
+    def consider_dangerous_change(self, _xfp):
+        pass
+
+
+unverified_fee_psbt = FakeOutputPSBT(external_input_psbt.fee_is_verified)
+psbtObject.consider_outputs(unverified_fee_psbt)
+assert unverified_fee_psbt.warnings[0][0] == 'Unverified Fee'
+assert all(label not in {'Big Fee', 'Huge Fee'} for label, _text in unverified_fee_psbt.warnings)
+
+verified_fee_psbt = FakeOutputPSBT(owned_input_psbt.fee_is_verified)
+psbtObject.consider_outputs(verified_fee_psbt)
+assert verified_fee_psbt.warnings[0][0] == 'Huge Fee'
+
+
+class FakeChain:
+    def render_value(self, value):
+        return str(value), 'sats'
+
+
+class FakeFlow:
+    chain = FakeChain()
+
+    def __init__(self, psbt):
+        self.psbt = psbt
+
+
+review = SignPsbtCommonFlow.render_warnings(FakeFlow(unverified_fee_psbt))
+assert 'Unverified' in review
+assert '4000 sats' not in review
+
+review = SignPsbtCommonFlow.render_warnings(FakeFlow(verified_fee_psbt))
+assert '4000 sats' in review
+
+return_value.write(b'OK')