EDGE: Harden PSBTv2 parsing and preserve Miniscript context

releases/EdgeChangeLog.md

@@ -13,6 +13,7 @@ This lists the changes in the most recent EDGE firmware, for each hardware platf
 
 # Shared Improvements - Both Mk4 and Q
 
+- Enhancement: Support per-input required height and time locktimes in PSBTv2 transactions.
 - New Feature: Added USB ncry v3 authenticated encryption with direction-separated keys and replay protection
 - Enhancement: Warn when a transaction's block-height `nLockTime` is more than
   ten years beyond the Bitcoin block height known to the firmware.
@@ -47,6 +48,10 @@ This lists the changes in the most recent EDGE firmware, for each hardware platf
   and mappings from a compromised USB host.
 - Bugfix: Fix device crash when message-signing input is valid JSON but not an
   object (NFC / QR / SD `.json` file). Thanks to [@Amiga500](https://github.com/Amiga500).
+- Bugfix: Harden PSBTv2 parsing by rejecting key data on singleton fields,
+  malformed global input/output count encodings, and files missing the required
+  global version.
+- Bugfix: Preserve Taproot context while deriving Miniscript policies.
 
 
 # Mk4 Specific Changes

releases/Next-ChangeLog.md

@@ -49,6 +49,10 @@ This lists the new changes that have not yet been published in a normal release.
   filesystems with more than one sector per cluster), fixing an integer underflow in
   `psram_copy_file`/`psram_mmap_file` that allowed out-of-bounds PSRAM writes, reads,
   and mappings from a compromised USB host.
+- Bugfix: Harden PSBTv2 parsing by rejecting key data on singleton fields,
+  malformed global input/output count encodings, and files missing the required
+  global version.
+- Bugfix: Preserve Taproot context while deriving Miniscript policies.
 
 # Mk Specific Changes
 

shared/miniscript.py

@@ -160,7 +160,7 @@ def derive(self, idx, key_map=None, change=False):
                 arg = arg.derive(idx, key_map, change)
 
             args.append(arg)
-        return type(self)(*args)
+        return type(self)(*args, taproot=self.taproot)
 
     @property
     def properties(self):

shared/psbt.py

@@ -448,7 +448,8 @@ def parse_subpaths(self, my_xfp, parent, cosign_xfp=None):
 # Track details of each output of PSBT
 #
 class psbtOutputProxy(psbtProxy):
-    no_keys = { PSBT_OUT_REDEEM_SCRIPT, PSBT_OUT_WITNESS_SCRIPT, PSBT_OUT_TAP_INTERNAL_KEY, PSBT_OUT_TAP_TREE }
+    no_keys = { PSBT_OUT_REDEEM_SCRIPT, PSBT_OUT_WITNESS_SCRIPT, PSBT_OUT_AMOUNT,
+                PSBT_OUT_SCRIPT, PSBT_OUT_TAP_INTERNAL_KEY, PSBT_OUT_TAP_TREE }
 
     blank_flds = ('unknown', 'subpaths', 'redeem_script', 'witness_script', 'sp_idxs',
                   'is_change', 'amount', 'script', 'attestation', 'proprietary',
@@ -693,7 +694,9 @@ class psbtInputProxy(psbtProxy):
     # only part-sigs have a key to be stored.
     no_keys = {PSBT_IN_NON_WITNESS_UTXO, PSBT_IN_WITNESS_UTXO, PSBT_IN_SIGHASH_TYPE,
                PSBT_IN_REDEEM_SCRIPT, PSBT_IN_WITNESS_SCRIPT, PSBT_IN_FINAL_SCRIPTSIG,
-               PSBT_IN_FINAL_SCRIPTWITNESS,PSBT_IN_TAP_KEY_SIG,
+               PSBT_IN_FINAL_SCRIPTWITNESS, PSBT_IN_PREVIOUS_TXID, PSBT_IN_OUTPUT_INDEX,
+               PSBT_IN_SEQUENCE, PSBT_IN_REQUIRED_TIME_LOCKTIME,
+               PSBT_IN_REQUIRED_HEIGHT_LOCKTIME, PSBT_IN_TAP_KEY_SIG,
                PSBT_IN_TAP_INTERNAL_KEY, PSBT_IN_TAP_MERKLE_ROOT}
 
     blank_flds = (
@@ -1383,7 +1386,10 @@ def serialize(self, out_fd, is_v2):
 class psbtObject(psbtProxy):
     "Just? parse and store"
     short_values = { PSBT_GLOBAL_TX_MODIFIABLE }
-    no_keys = { PSBT_GLOBAL_UNSIGNED_TX, PSBT_GLOBAL_GENERIC_SIGNED_MESSAGE }
+    no_keys = { PSBT_GLOBAL_UNSIGNED_TX, PSBT_GLOBAL_TX_VERSION,
+                PSBT_GLOBAL_FALLBACK_LOCKTIME, PSBT_GLOBAL_INPUT_COUNT,
+                PSBT_GLOBAL_OUTPUT_COUNT, PSBT_GLOBAL_TX_MODIFIABLE,
+                PSBT_GLOBAL_VERSION, PSBT_GLOBAL_GENERIC_SIGNED_MESSAGE }
     blank_flds = ("hashPrevouts", "hashSequence", "hashOutputs", "hashValues", "hashScriptPubKeys",
                   "tr_hashPrevouts", "tr_hashSequence", "tr_hashOutputs", "my_tr_in", "unknown")
 
@@ -1499,10 +1505,14 @@ def store(self, kt, key, val):
         elif kt == PSBT_GLOBAL_FALLBACK_LOCKTIME:
             self.fallback_locktime = unpack("<I", self.get(val))[0]
         elif kt == PSBT_GLOBAL_INPUT_COUNT:
-            self.num_inputs = deser_compact_size(BytesIO(self.get(val)))
+            raw = self.get(val)
+            self.num_inputs = deser_compact_size(BytesIO(raw))
+            assert raw == ser_compact_size(self.num_inputs), "invalid input count"
             self.has_gic = True
         elif kt == PSBT_GLOBAL_OUTPUT_COUNT:
-            self.num_outputs = deser_compact_size(BytesIO(self.get(val)))
+            raw = self.get(val)
+            self.num_outputs = deser_compact_size(BytesIO(raw))
+            assert raw == ser_compact_size(self.num_outputs), "invalid output count"
             self.has_goc = True
         elif kt == PSBT_GLOBAL_TX_MODIFIABLE:
             # bytes of length 1 (tx modifiable in short_values)
@@ -1816,9 +1826,9 @@ async def validate(self):
             # verision is provided in PSBT - take it as given
             assert self.version in (0,2)
         else:
-            # PSBT version is not defined
-            # global unsigned tx is only allowed in v0
-            self.version = 2 if self.txn is None else 0
+            # PSBTv0 may omit its version, but PSBTv2 must specify version 2.
+            assert self.txn, "v2 requires global version"
+            self.version = 0
 
         self.is_v2 = self.version is not None and self.version >= 2
 

testing/test_sign.py

@@ -7,7 +7,13 @@
 from ckcc_protocol.protocol import CCProtocolPacker, CCProtoError
 from binascii import b2a_hex, a2b_hex
 from psbt import BasicPSBT, BasicPSBTInput, BasicPSBTOutput, PSBT_IN_REDEEM_SCRIPT
-from psbt import PSBT_GLOBAL_VERSION, PSBT_IN_WITNESS_UTXO, PSBT_OUT_AMOUNT
+from psbt import (PSBT_GLOBAL_VERSION, PSBT_GLOBAL_TX_VERSION,
+                  PSBT_GLOBAL_FALLBACK_LOCKTIME, PSBT_GLOBAL_INPUT_COUNT,
+                  PSBT_GLOBAL_OUTPUT_COUNT, PSBT_GLOBAL_TX_MODIFIABLE,
+                  PSBT_IN_WITNESS_UTXO, PSBT_IN_PREVIOUS_TXID,
+                  PSBT_IN_OUTPUT_INDEX, PSBT_IN_SEQUENCE,
+                  PSBT_IN_REQUIRED_TIME_LOCKTIME, PSBT_IN_REQUIRED_HEIGHT_LOCKTIME,
+                  PSBT_OUT_AMOUNT, PSBT_OUT_SCRIPT)
 from io import BytesIO
 from pprint import pprint
 from decimal import Decimal
@@ -83,6 +89,65 @@ def add_duplicate(psbt):
 
     assert 'PSBT parse failed' in ee.value.args[0]
 
+
+@pytest.mark.parametrize('scope, ktype, value', [
+    ('global', PSBT_GLOBAL_TX_VERSION, struct.pack('<I', 2)),
+    ('global', PSBT_GLOBAL_FALLBACK_LOCKTIME, struct.pack('<I', 0)),
+    ('global', PSBT_GLOBAL_INPUT_COUNT, b'\x01'),
+    ('global', PSBT_GLOBAL_OUTPUT_COUNT, b'\x01'),
+    ('global', PSBT_GLOBAL_TX_MODIFIABLE, b'\x00'),
+    ('global', PSBT_GLOBAL_VERSION, struct.pack('<I', 2)),
+    ('input', PSBT_IN_PREVIOUS_TXID, bytes(32)),
+    ('input', PSBT_IN_OUTPUT_INDEX, struct.pack('<I', 0)),
+    ('input', PSBT_IN_SEQUENCE, struct.pack('<I', 0xffffffff)),
+    ('input', PSBT_IN_REQUIRED_TIME_LOCKTIME, struct.pack('<I', 500000000)),
+    ('input', PSBT_IN_REQUIRED_HEIGHT_LOCKTIME, struct.pack('<I', 1)),
+    ('output', PSBT_OUT_AMOUNT, struct.pack('<q', 1000)),
+    ('output', PSBT_OUT_SCRIPT, b'\x51'),
+])
+def test_psbt_singleton_rejects_key_data(try_sign, fake_txn, scope, ktype, value):
+    def add_key_data(psbt):
+        target = psbt
+        if scope == 'input':
+            target = psbt.inputs[0]
+        elif scope == 'output':
+            target = psbt.outputs[0]
+        target.unknown = [(bytes([ktype, 0]), value)]
+
+    psbt = fake_txn(1, 1, psbt_v2=True, psbt_hacker=add_key_data)
+
+    with pytest.raises(CCProtoError) as ee:
+        try_sign(psbt, accept=False)
+
+    assert 'PSBT parse failed' in ee.value.args[0]
+
+
+@pytest.mark.parametrize('ktype', [PSBT_GLOBAL_INPUT_COUNT, PSBT_GLOBAL_OUTPUT_COUNT])
+def test_psbt_v2_rejects_non_exact_global_count(try_sign, fake_txn, ktype):
+    def add_trailing_count_byte(psbt):
+        if ktype == PSBT_GLOBAL_INPUT_COUNT:
+            psbt.input_count = None
+        else:
+            psbt.output_count = None
+        psbt.unknown = [(bytes([ktype]), b'\x01\x00')]
+
+    psbt = fake_txn(1, 1, psbt_v2=True, psbt_hacker=add_trailing_count_byte)
+
+    with pytest.raises(CCProtoError) as ee:
+        try_sign(psbt, accept=False)
+
+    assert 'PSBT parse failed' in ee.value.args[0]
+
+
+def test_psbt_v2_requires_global_version(try_sign, fake_txn):
+    psbt = fake_txn(1, 1, psbt_v2=True,
+                    psbt_hacker=lambda p: setattr(p, 'version', None))
+
+    with pytest.raises(CCProtoError) as ee:
+        try_sign(psbt, accept=False)
+
+    assert 'Invalid PSBT' in ee.value.args[0]
+
 @pytest.mark.parametrize('fn', [
 	'data/2-of-2.psbt',
 	'data/filled_scriptsig.psbt',

testing/test_unit.py

@@ -183,6 +183,14 @@ def test_tapscript_leaf_version(sim_exec):
     )
     assert "Tapleaf ver 0xc1" in rv
 
+
+def test_miniscript_derive_preserves_taproot(sim_exec):
+    rv = sim_exec(
+        "from miniscript import Miniscript; "
+        "RV.write(str(Miniscript(taproot=True).derive(0).taproot))"
+    )
+    assert rv == "True"
+
 @pytest.mark.parametrize('secret,counter,expect', [
         ( b'abcdefghij', 1, '765705'),
         ( b'abcdefghij', 2, '816065'),