Cut repeated computation out of PSBTParser.parse
What changed, and why it matters
This commit is a performance optimization, not a security fix. It speeds up parsing of Bitcoin transaction files (PSBTs) on the slow Pi Zero hardware by avoiding repeated work: reading the transaction once instead of rebuilding it twelve times, using a direct fingerprint lookup instead of deriving a child key, and caching intermediate key derivations during parsing. The author explicitly states the parse output is unchanged and adds tests to prove the cached and uncached paths produce identical results.
No security action required. Treat as a normal performance refactor; review for correctness during routine code review.
Security signals we found
No strong security signals were identified.
Evidence from the diff
The change refactors PSBTParser.parse() and helpers to reduce redundant computation. Key changes: (1) store self.psbt.tx.vout once before the output loop; (2) replace self.root.child(0).fingerprint with self.root.my_fingerprint in _fill_missing_fingerprints; (3) introduce _derive_with_cache() to memoize HDKey.child() results keyed by (id(parent_key), derivation_path_so_far), with a MAX_CACHED_DERIVATIONS bound of 1000, and clear the cache in a finally block after each parse. The cache is threaded through _get_policy, _get_cosigners, _parse_outputs, and _fill_missing_fingerprints. Tests verify byte-identical output with and without the cache across mainnet/testnet fixtures.
Changed components
src/seedsigner/models/psbt_parser.pytests/test_psbt_parser.pyInspect captured patch +331 / −34
### src/seedsigner/models/psbt_parser.py
@@ -20,6 +20,10 @@ class OPCODES:
class PSBTParser():
+ # Upper bound on how many derived levels a single parse will hold on to; see
+ # _derive_with_cache.
+ MAX_CACHED_DERIVATIONS = 1000
+
def __init__(self, p: PSBT, seed: Seed, network: str = SettingsConstants.MAINNET):
self.psbt: PSBT = p
self.seed = seed
@@ -37,6 +41,7 @@ def __init__(self, p: PSBT, seed: Seed, network: str = SettingsConstants.MAINNET
self.op_return_data: bytes = None
self.root = None
+ self._child_key_derivation_cache = {}
if self.seed is not None:
self.parse()
@@ -70,6 +75,24 @@ def _set_root(self):
def parse(self):
+ """
+ Parsing traverses a derivation path down to an individual address one level at a
+ time, over and over, and where that traversal begins depends on the wallet.
+
+ Single-sig traverses the full path down from our own master key, on every OUTPUT
+ the PSBT claims is ours.
+
+ Multisig instead traverses just the last two levels down from each cosigner's
+ account xpub — once per cosigner, on every INPUT and on every OUTPUT carrying the
+ multisig script.
+
+ Deriving each level costs a hash and an elliptic curve operation, and these
+ traversals overlap heavily: everything in one account shares the same opening
+ levels, differing only in the address at the end.
+
+ So every level derived during this parse is kept in _child_key_derivation_cache
+ and reused. See _derive_with_cache.
+ """
if self.psbt is None:
logger.info(f"self.psbt is None!!")
return False
@@ -80,18 +103,26 @@ def parse(self):
self._set_root()
- # Try to fix missing fingerprints before parsing
- self._fill_missing_fingerprints()
+ self._child_key_derivation_cache = {}
- rt = self._parse_inputs()
- if rt == False:
- return False
+ try:
+ # Try to fix missing fingerprints before parsing
+ self._fill_missing_fingerprints()
- rt = self._parse_outputs()
- if rt == False:
- return False
+ rt = self._parse_inputs()
+ if rt == False:
+ return False
- return True
+ rt = self._parse_outputs()
+ if rt == False:
+ return False
+
+ return True
+ finally:
+ # The cache is only useful within a single parse and it holds keys derived
+ # from the signing seed, so drop it now rather than letting it live on for
+ # as long as this parser does.
+ self._child_key_derivation_cache = {}
def _parse_inputs(self):
@@ -105,7 +136,7 @@ def _parse_inputs(self):
self.input_amount += inp.utxo.value
script_pubkey = inp.script_pubkey
- inp_policy = PSBTParser._get_policy(inp, script_pubkey, self.psbt.xpubs)
+ inp_policy = PSBTParser._get_policy(inp, script_pubkey, self.psbt.xpubs, self._child_key_derivation_cache)
if self.policy == None:
self.policy = inp_policy
else:
@@ -119,8 +150,14 @@ def _parse_outputs(self):
self.fee_amount = 0
self.destination_addresses = []
self.destination_amounts = []
+
+ # Asking the PSBT for its transaction rebuilds that entire transaction from
+ # scratch on every single request. The outputs are consulted a dozen times
+ # over the course of the loop below, so grab them once now.
+ vout = self.psbt.tx.vout
+
for i, out in enumerate(self.psbt.outputs):
- out_policy = PSBTParser._get_policy(out, self.psbt.tx.vout[i].script_pubkey, self.psbt.xpubs)
+ out_policy = PSBTParser._get_policy(out, vout[i].script_pubkey, self.psbt.xpubs, self._child_key_derivation_cache)
is_change = False
# if policy is the same - probably change
@@ -157,7 +194,7 @@ def _parse_outputs(self):
# should be one or zero for single-key addresses
if len(out.bip32_derivations.values()) > 0:
der = list(out.bip32_derivations.values())[0].derivation
- my_pubkey = self.root.derive(der)
+ my_pubkey = PSBTParser._derive_with_cache(self.root, der, self._child_key_derivation_cache)
if self.policy["type"] == "p2pkh" and my_pubkey is not None:
sc = script.p2pkh(my_pubkey)
@@ -168,7 +205,7 @@ def _parse_outputs(self):
elif self.policy["type"] == "p2wpkh" and my_pubkey is not None:
sc = script.p2wpkh(my_pubkey)
- if sc.data == self.psbt.tx.vout[i].script_pubkey.data:
+ if sc.data == vout[i].script_pubkey.data:
is_change = True
elif "p2tr" in self.policy["type"]:
@@ -178,21 +215,21 @@ def _parse_outputs(self):
# TODO: Support keys in taptree leaves
leaf_hashes, derivation = list(out.taproot_bip32_derivations.values())[0]
der = derivation.derivation
- my_pubkey = self.root.derive(der)
+ my_pubkey = PSBTParser._derive_with_cache(self.root, der, self._child_key_derivation_cache)
sc = script.p2tr(my_pubkey)
- if sc.data == self.psbt.tx.vout[i].script_pubkey.data:
+ if sc.data == vout[i].script_pubkey.data:
is_change = True
- if sc.data == self.psbt.tx.vout[i].script_pubkey.data:
+ if sc.data == vout[i].script_pubkey.data:
is_change = True
- if self.psbt.tx.vout[i].script_pubkey.data[0] == OPCODES.OP_RETURN:
+ if vout[i].script_pubkey.data[0] == OPCODES.OP_RETURN:
# The data is written as: OP_RETURN + OP_PUSHDATA1 + len(payload) + payload
- self.op_return_data = self.psbt.tx.vout[i].script_pubkey.data[3:]
+ self.op_return_data = vout[i].script_pubkey.data[3:]
elif is_change:
- addr = self.psbt.tx.vout[i].script_pubkey.address(NETWORKS[SettingsConstants.map_network_to_embit(self.network)])
+ addr = vout[i].script_pubkey.address(NETWORKS[SettingsConstants.map_network_to_embit(self.network)])
fingerprints = []
derivation_paths = []
@@ -211,17 +248,17 @@ def _parse_outputs(self):
self.change_data.append({
"output_index": i,
"address": addr,
- "amount": self.psbt.tx.vout[i].value,
+ "amount": vout[i].value,
"fingerprint": fingerprints,
"derivation_path": derivation_paths,
})
- self.change_amount += self.psbt.tx.vout[i].value
+ self.change_amount += vout[i].value
else:
- addr = self.psbt.tx.vout[i].script_pubkey.address(NETWORKS[SettingsConstants.map_network_to_embit(self.network)])
+ addr = vout[i].script_pubkey.address(NETWORKS[SettingsConstants.map_network_to_embit(self.network)])
self.destination_addresses.append(addr)
- self.destination_amounts.append(self.psbt.tx.vout[i].value)
- self.spend_amount += self.psbt.tx.vout[i].value
+ self.destination_amounts.append(vout[i].value)
+ self.spend_amount += vout[i].value
self.fee_amount = self.psbt.fee()
return True
@@ -256,7 +293,7 @@ def sig_count(tx):
@staticmethod
- def _get_policy(scope, scriptpubkey, xpubs):
+ def _get_policy(scope, scriptpubkey, xpubs, child_key_derivation_cache=None):
"""Parse scope and get policy"""
# we don't know the policy yet, let's parse it
script_type = scriptpubkey.script_type()
@@ -286,7 +323,7 @@ def _get_policy(scope, scriptpubkey, xpubs):
# check pubkeys are derived from cosigners
try:
- cosigners = PSBTParser._get_cosigners(pubkeys, scope.bip32_derivations, xpubs)
+ cosigners = PSBTParser._get_cosigners(pubkeys, scope.bip32_derivations, xpubs, child_key_derivation_cache)
policy.update({"m": m, "n": n, "cosigners": cosigners})
except:
policy.update({"m": m, "n": n})
@@ -324,7 +361,53 @@ def _parse_multisig(sc):
@staticmethod
- def _get_cosigners(pubkeys, derivations, xpubs):
+ def _derive_with_cache(parent_key: bip32.HDKey, derivation_path: List[int], cache: dict | None = None) -> bip32.HDKey:
+ """
+ Derives the key that sits at the given derivation path below parent_key, reusing
+ any levels along the way that have already been derived during this parse.
+
+ A derivation path is traversed one level at a time, and two derivation paths that
+ begin the same way share those opening levels. Each level reached is stored in the
+ cache, so a later derivation running through that level picks it up instead of
+ deriving it a second time.
+
+ Entries are keyed on (id(parent_key), derivation_path_so_far) — the path traversed
+ down from that parent to reach this point. id() is the Python built-in for an
+ object's identity; the parent belongs in the key because a multisig parse runs
+ these same derivations below each cosigner's xpub in turn.
+
+ Keying on the parent's fingerprint was rejected: four bytes is small enough for a
+ malicious coordinator to grind a deliberate collision, and the cosigner xpubs come
+ from the psbt.
+
+ The cache stops accepting new levels at MAX_CACHED_DERIVATIONS. A real wallet
+ needs orders of magnitude fewer, but a hostile psbt can name any number of
+ derivation paths, and every level of every one of them would otherwise be held
+ in memory at once. Past the limit each level is simply derived as needed.
+ """
+ if cache is None:
+ return parent_key.derive(derivation_path)
+
+ derived_key = parent_key
+ derivation_path_so_far = ()
+
+ # Traverse the derivation path...
+ for index in derivation_path:
+ derivation_path_so_far += (index,)
+ cache_key = (id(parent_key), derivation_path_so_far)
+ already_derived = cache.get(cache_key)
+ if already_derived is None:
+ # First time deriving this level. Do the work to derive this level's child
+ # and store it in the cache.
+ already_derived = derived_key.child(index)
+ if len(cache) < PSBTParser.MAX_CACHED_DERIVATIONS:
+ cache[cache_key] = already_derived
+ derived_key = already_derived
+ return derived_key
+
+
+ @staticmethod
+ def _get_cosigners(pubkeys, derivations, xpubs, child_key_derivation_cache=None):
"""Returns xpubs used to derive pubkeys using global xpub field from psbt"""
cosigners = []
for i, pubkey in enumerate(pubkeys):
@@ -338,7 +421,9 @@ def _get_cosigners(pubkeys, derivations, xpubs):
# check derivation - last two indexes give pub from xpub
if origin_der.derivation == der.derivation[:-2]:
# check that it derives to pubkey actually
- if xpub.derive(der.derivation[-2:]).key == pubkey:
+ derived_key = PSBTParser._derive_with_cache(
+ xpub, der.derivation[-2:], child_key_derivation_cache)
+ if derived_key.key == pubkey:
# append strings so they can be sorted and compared
cosigners.append(xpub.to_base58())
break
@@ -431,11 +516,10 @@ def _fill_missing_fingerprints(self):
"""
if not self.root:
return 0
-
+
def _fill_scope(scope: InputScope | OutputScope):
"""Helper function to fill missing fingerprints in a scope (input/output)"""
- signing_seed_fingerprint = self.root.child(0).fingerprint
-
+
# Helper function to check and fix fingerprint
def _get_updated_fingerprint(public_key: PublicKey, derivation_path_obj: DerivationPath) -> DerivationPath | None:
if derivation_path_obj.fingerprint != b"\x00\x00\x00\x00":
@@ -447,9 +531,10 @@ def _get_updated_fingerprint(public_key: PublicKey, derivation_path_obj: Derivat
# is owned by the signing seed. In that case we populate the missing (zero)
# fingerprint with the signing seed's master fingerprint so downstream
# parsing/signing can treat it as owned by this seed.
- derived_key = self.root.derive(derivation_path_obj.derivation)
+ derived_key = PSBTParser._derive_with_cache(
+ self.root, derivation_path_obj.derivation, self._child_key_derivation_cache)
if derived_key.key.sec() == public_key.sec():
- return DerivationPath(signing_seed_fingerprint, derivation_path_obj.derivation)
+ return DerivationPath(self.root.my_fingerprint, derivation_path_obj.derivation)
return None
# Handle regular BIP32 derivations
### tests/test_psbt_parser.py
@@ -2,8 +2,10 @@
import random
from binascii import a2b_base64
+from copy import deepcopy
from embit import bip32
-from embit.psbt import PSBT
+from embit.networks import NETWORKS
+from embit.psbt import PSBT, DerivationPath
from embit.descriptor import Descriptor
from seedsigner.models.psbt_parser import PSBTParser
@@ -485,3 +487,213 @@ def test_parse_op_return_content():
assert psbt_parser.change_amount == 99992296
assert psbt_parser.destination_addresses == []
assert psbt_parser.destination_amounts == []
+
+
+
+class TestPSBTParserOptimizations:
+ """
+ Guard tests for the parse-time optimizations in PSBTParser.
+
+ These verify the claims the speedups rely on:
+ * root.my_fingerprint equals root.child(0).fingerprint
+ * reusing an already-derived level yields exactly the same key as deriving it
+ again.
+ """
+ seed = PSBTTestData.seed
+
+ def _root(self, seed: Seed = None) -> bip32.HDKey:
+ if seed is None:
+ seed = self.seed
+ return bip32.HDKey.from_seed(
+ seed.seed_bytes, version=NETWORKS["main"]["xprv"])
+
+
+ def test_my_fingerprint_equals_child0_fingerprint(self):
+ """
+ Reading my_fingerprint in place of child(0).fingerprint is byte-identical,
+ because HDKey.child(0) sets its .fingerprint to hash160(parent.sec())[:4],
+ which is exactly parent.my_fingerprint.
+
+ This is really a unit test / regression test against embit itself, but it is worth
+ testing here.
+ """
+ root = self._root()
+ assert root.my_fingerprint == root.child(0).fingerprint
+
+
+ def test_zero_fingerprint_fill_over_many_inputs(self, monkeypatch):
+ """
+ The inputs in this test have their fingerprints blanked (set to all zero), which
+ should then require one full derivation per input to work out whether that input
+ is ours.
+
+ The artificial inputs in this test share the same full derivation path so each
+ level should only be derived once total rather than once per input.
+ """
+ psbt = PSBT.parse(a2b_base64(PSBTTestData.SINGLE_SIG_NATIVE_SEGWIT_1_INPUT))
+ master_fingerprint = self._root().my_fingerprint
+
+ # Sanity check that this artificial psbt has no outputs. We have to make sure that
+ # the derivation counts at the end of the test were only for inputs, not outputs.
+ assert len(psbt.outputs) == 0, "fixture is expected to have no outputs"
+
+ # Artificially boost this test psbt to 10 total inputs from the same wallet
+ for _ in range(9):
+ psbt.inputs.append(deepcopy(psbt.inputs[0]))
+
+ # Zero out all of the inputs' fingerprints
+ num_zeroed = 0
+ for inp in psbt.inputs:
+ for pub, dp in list(inp.bip32_derivations.items()):
+ inp.bip32_derivations[pub] = DerivationPath(b"\x00\x00\x00\x00", dp.derivation)
+ num_zeroed += 1
+ assert num_zeroed == len(psbt.inputs), "fixture did not yield one derivation per input"
+
+ num_levels = len(list(psbt.inputs[0].bip32_derivations.values())[0].derivation)
+
+ # Attach a counter to track every level actually derived during the parse
+ num_derivations = 0
+ uncounted_child = bip32.HDKey.child
+ def counting_child(self, index, hardened=False):
+ nonlocal num_derivations # reference the above var outside the function scope
+ num_derivations += 1
+ return uncounted_child(self, index, hardened)
+ monkeypatch.setattr(bip32.HDKey, "child", counting_child)
+
+ # Instantiating the parser with the psbt will automatically fill in the zeroed
+ # fingerprints.
+ PSBTParser(psbt, self.seed, network=SettingsConstants.MAINNET)
+
+ # All 10 inputs share the one derivation path, so each of its levels should have
+ # been derived exactly once between them, rather than once per input.
+ assert num_derivations == num_levels
+
+ # Sanity check: num_derivations could be correct when just ONE of the ten inputs
+ # was processed. Confirm that EVERY input really was processed by verifying that
+ # each input was filled in with the correct fingerprint.
+ for inp in psbt.inputs:
+ for pub, dp in inp.bip32_derivations.items():
+ assert dp.fingerprint == master_fingerprint
+
+
+ def test_derive_with_cache_matches_plain_derive(self):
+ """A cached traversal down a derivation path must land on exactly the same key
+ as an uncached one, whether or not earlier derivation paths already populated
+ the cache."""
+ root = self._root()
+ derivation_paths = [
+ [84 + 0x80000000, 1 + 0x80000000, 0x80000000, 0, 0],
+ [84 + 0x80000000, 1 + 0x80000000, 0x80000000, 0, 1], # shares 4 levels
+ [84 + 0x80000000, 1 + 0x80000000, 0x80000000, 1, 0], # shares 3 levels
+ [48 + 0x80000000, 0x80000000, 0x80000000, 2 + 0x80000000], # different account
+ ]
+ cache = {}
+ for derivation_path in derivation_paths:
+ cached = PSBTParser._derive_with_cache(root, derivation_path, cache)
+ assert cached.key.sec() == root.derive(derivation_path).key.sec()
+ # and with no cache supplied at all
+ uncached = PSBTParser._derive_with_cache(root, derivation_path)
+ assert uncached.key.sec() == root.derive(derivation_path).key.sec()
+
+ # the shared opening levels were derived once, not once per derivation path
+ assert len(cache) == 5 + 1 + 2 + 4
+
+
+ def test_derive_with_cache_does_not_cross_parent_keys(self):
+ """
+ Multisig traverses the same relative derivation path below every cosigner's
+ account xpub. Verify that the cache properly keeps the parents' cache data
+ separate despite having derivations that share the same relative path.
+ """
+ # Two cosigners' account xpubs from the multisig test fixtures
+ cosigner_a_xpub = self._root(PSBTTestData.multisig_key_2).derive("m/48h/0h/0h/2h").to_public()
+ cosigner_b_xpub = self._root(PSBTTestData.multisig_key_3).derive("m/48h/0h/0h/2h").to_public()
+
+ # The receive address at index 5 is: m/48h/0h/0h/2h/0/5. The parent xpubs already
+ # have the first 4 levels derived, so this operation is only the final two levels.
+ receive_index_5 = [0, 5]
+ cache = {}
+
+ # The cache here isn't providing any speedup (there are no derivations in the
+ # cache to take advantage of), but we're just testing that the cache doesn't
+ # confuse/combine the two cosigners' derivation data.
+ from_a = PSBTParser._derive_with_cache(cosigner_a_xpub, receive_index_5, cache)
+ from_b = PSBTParser._derive_with_cache(cosigner_b_xpub, receive_index_5, cache)
+
+ assert len(cache) == 4, "the cache should have 4 entries: 2 levels for each cosigner's parent xpub"
+
+ assert from_a.key.sec() != from_b.key.sec()
+
+ # The result derived with the cache must be identical to deriving from the xpub
+ # directly.
+ assert from_a.key.sec() == cosigner_a_xpub.derive(receive_index_5).key.sec()
+ assert from_b.key.sec() == cosigner_b_xpub.derive(receive_index_5).key.sec()
+
+
+ def test_get_cosigners_identical_with_and_without_cache(self):
+ """The cache is transparent to callers: _get_cosigners returns the same cosigner
+ list whether or not a cache is threaded in."""
+ psbt = PSBT.parse(a2b_base64(PSBTTestData.MULTISIG_NATIVE_SEGWIT_1_INPUT))
+ inp = psbt.inputs[0]
+ pubkeys = list(inp.bip32_derivations.keys())
+
+ uncached = PSBTParser._get_cosigners(pubkeys, inp.bip32_derivations, psbt.xpubs)
+
+ child_key_derivation_cache = {}
+ cached = PSBTParser._get_cosigners(
+ pubkeys, inp.bip32_derivations, psbt.xpubs, child_key_derivation_cache)
+
+ assert cached == uncached
+ assert len(child_key_derivation_cache) > 0, "the cache was never populated"
+
+
+ def test_cache_does_not_change_parse_output(self, monkeypatch):
+ """
+ The whole point of the cache is that it changes nothing at all. Parse a spread of
+ psbts twice — once normally, once with every cache lookup forced to miss — and
+ require identical parser state and identical resulting psbt bytes.
+
+ The fill path rewrites fingerprints into the psbt itself, so the serialized psbt
+ is compared too, not just the parser's own attributes.
+ """
+ cases = [
+ (PSBTTestData.SINGLE_SIG_NATIVE_SEGWIT_1_INPUT, PSBTTestData.SINGLE_SIG_NATIVE_SEGWIT_CHANGE),
+ (PSBTTestData.SINGLE_SIG_TAPROOT_1_INPUT, PSBTTestData.SINGLE_SIG_TAPROOT_CHANGE),
+ (PSBTTestData.MULTISIG_NATIVE_SEGWIT_1_INPUT, PSBTTestData.MULTISIG_NATIVE_SEGWIT_CHANGE),
+ (PSBTTestData.MULTISIG_LEGACY_P2SH_1_INPUT, PSBTTestData.MULTISIG_LEGACY_P2SH_CHANGE),
+ ]
+
+ def parse_everything():
+ results = []
+ for input_base64, change_hex in cases:
+ for zero_fingerprints in (False, True):
+ psbt = PSBT.parse(a2b_base64(input_base64))
+ psbt.outputs.append(create_output(change_hex, 10_000))
+ if zero_fingerprints:
+ for scope in list(psbt.inputs) + list(psbt.outputs):
+ for pub, dp in list(scope.bip32_derivations.items()):
+ scope.bip32_derivations[pub] = DerivationPath(
+ b"\x00\x00\x00\x00", dp.derivation)
+ parser = PSBTParser(psbt, self.seed, network=SettingsConstants.REGTEST)
+ results.append((
+ repr(parser.policy),
+ parser.input_amount,
+ parser.spend_amount,
+ parser.change_amount,
+ parser.fee_amount,
+ parser.destination_addresses,
+ parser.destination_amounts,
+ repr(parser.change_data),
+ parser.op_return_data,
+ psbt.serialize(),
+ ))
+ return results
+
+ with_cache = parse_everything()
+
+ # Hand every call its own throwaway cache so no lookup can ever hit
+ uncached = PSBTParser._derive_with_cache
+ monkeypatch.setattr(PSBTParser, "_derive_with_cache", staticmethod(
+ lambda parent_key, derivation_path, cache=None: uncached(parent_key, derivation_path)))
+
+ assert parse_everything() == with_cacheWhy this scored 15/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.