Merge pull request #1002 from kdmukai/psbt_parser_derivation_cache
What changed, and why it matters
This commit is a performance improvement for parsing Bitcoin transaction files (PSBTs). It adds a cache so the wallet doesn't repeatedly recalculate the same cryptographic key derivations, and it avoids rebuilding the whole transaction object many times. The change also removes a small block of code that handled an older multisig script type (p2sh) when deciding whether an output is change. The commit message and code comments describe this only as a performance optimization, not a security fix.
No immediate security action is required. Treat this as a routine performance refactor. If the removed p2sh change-detection path is still relevant to supported wallet policies, verify through tests or release notes that legacy p2sh multisig change outputs are still handled correctly elsewhere. Reviewers may also want to confirm that id(parent_key) reuse concerns are acceptable given the parent reference is kept alive in each cache entry.
Security signals we found
Removed p2sh redeem-script handling path in change detection
New derivation cache keyed by id(parent_key) with explicit rejection of fingerprint-based keying due to collision risk
Fingerprint recovery now uses root.my_fingerprint instead of root.child(0).fingerprint
Cache capped at MAX_CACHED_DERIVATIONS to bound memory growth
Evidence from the diff
The patch refactors PSBTParser.parse() to pass a child_key_derivation_cache dict through _parse_inputs(), _parse_outputs(), _get_policy(), _get_cosigners(), and _fill_missing_fingerprints(). A new static method _derive_with_cache() memoizes intermediate HDKey.child() results keyed by (id(parent_key), derivation_path_so_far), with a MAX_CACHED_DERIVATIONS cap of 1000. It also stores vout = self.psbt.tx.vout once instead of repeatedly accessing self.psbt.tx.vout[i]. A behavioral change is the removal of the p2sh redeem_script branch when reconstructing change scripts in _parse_outputs(); only p2wsh, p2wsh-p2sh, p2pkh, p2sh-p2wpkh, p2wpkh, and p2tr paths remain. The fingerprint-fill helper now uses self.root.my_fingerprint instead of self.root.child(0).fingerprint (documented as equivalent). Tests verify cache correctness, parent isolation, and that capped cache does not alter parse output.
Changed components
src/seedsigner/models/psbt_parser.pytests/test_psbt_parser.pyInspect captured patch +419 / −35
### src/seedsigner/models/psbt_parser.py
@@ -20,6 +20,26 @@ class OPCODES:
class PSBTParser():
+ """
+ Reads a psbt on behalf of one seed and works out everything the signing flow shows
+ the user before they approve: the wallet policy (script type, plus m-of-n and the
+ cosigners for multisig), the amount coming in, what is being spent, what comes back
+ as change, the fee, where the spend is going, and any OP_RETURN payload.
+
+ Constructing it with a seed parses immediately. The results are read off the instance
+ attributes, with per-change-output detail in change_data.
+
+ has_matching_input_fingerprint answers the earlier question of which seed a psbt is
+ for and needs no parse.
+ """
+
+ # Upper bound on how many levels of derivation a single parse will cache. 1000 is
+ # just slightly under a 3-of-5 multisig consolidating 200 inputs, which costs roughly
+ # 650 kilobytes. A psbt that needs more levels than that still parses correctly; it
+ # just stops getting cache hits once the cache is full.
+ MAX_CACHED_DERIVATIONS = 1000
+
+
def __init__(self, p: PSBT, seed: Seed, network: str = SettingsConstants.MAINNET):
self.psbt: PSBT = p
self.seed = seed
@@ -70,6 +90,26 @@ 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 a cache and reused. See
+ _derive_with_cache.
+
+ Note that the cache is only useful within a single parse so it is not preserved.
+ """
if self.psbt is None:
logger.info(f"self.psbt is None!!")
return False
@@ -80,21 +120,23 @@ def parse(self):
self._set_root()
+ child_key_derivation_cache = {}
+
# Try to fix missing fingerprints before parsing
- self._fill_missing_fingerprints()
+ self._fill_missing_fingerprints(child_key_derivation_cache)
- rt = self._parse_inputs()
+ rt = self._parse_inputs(child_key_derivation_cache)
if rt == False:
return False
- rt = self._parse_outputs()
+ rt = self._parse_outputs(child_key_derivation_cache)
if rt == False:
return False
return True
- def _parse_inputs(self):
+ def _parse_inputs(self, child_key_derivation_cache: dict):
self.input_amount = 0
self.num_inputs = len(self.psbt.inputs)
for inp in self.psbt.inputs:
@@ -105,22 +147,28 @@ 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, child_key_derivation_cache)
if self.policy == None:
self.policy = inp_policy
else:
if self.policy != inp_policy:
raise RuntimeError("Mixed inputs in the transaction")
- def _parse_outputs(self):
+ def _parse_outputs(self, child_key_derivation_cache: dict):
self.spend_amount = 0
self.change_amount = 0
self.change_data = []
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, child_key_derivation_cache)
is_change = False
# if policy is the same - probably change
@@ -135,10 +183,6 @@ def _parse_outputs(self):
# empty script by default
sc = script.Script(b"")
- # if older multisig, just use existing script
- if self.policy["type"] == "p2sh":
- sc = script.p2sh(out.redeem_script)
-
# multisig, we know witness script
if self.policy["type"] == "p2wsh":
sc = script.p2wsh(out.witness_script)
@@ -157,7 +201,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, child_key_derivation_cache)
if self.policy["type"] == "p2pkh" and my_pubkey is not None:
sc = script.p2pkh(my_pubkey)
@@ -168,7 +212,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 +222,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, 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 +255,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 +300,7 @@ def sig_count(tx):
@staticmethod
- def _get_policy(scope, scriptpubkey, xpubs):
+ def _get_policy(scope, scriptpubkey, xpubs, child_key_derivation_cache: dict | None):
"""Parse scope and get policy"""
# we don't know the policy yet, let's parse it
script_type = scriptpubkey.script_type()
@@ -286,7 +330,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 +368,58 @@ def _parse_multisig(sc):
@staticmethod
- def _get_cosigners(pubkeys, derivations, xpubs):
+ def _derive_with_cache(parent_key: bip32.HDKey, derivation_path: List[int], child_key_derivation_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.
+
+ Each entry also holds on to the parent it was derived from. id() is only the
+ object's address, which Python is free to hand to a new object once the original
+ is released. Keeping the parent means its address cannot be reused for as long as
+ the entry it belongs to is alive.
+
+ 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.
+ """
+ if child_key_derivation_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)
+ cached_entry = child_key_derivation_cache.get(cache_key)
+ if cached_entry 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(child_key_derivation_cache) < PSBTParser.MAX_CACHED_DERIVATIONS:
+ # Parent must also be stored to keep its id() from being reused
+ child_key_derivation_cache[cache_key] = (parent_key, already_derived)
+ else:
+ cached_parent, already_derived = cached_entry
+ derived_key = already_derived
+ return derived_key
+
+
+ @staticmethod
+ def _get_cosigners(pubkeys, derivations, xpubs, child_key_derivation_cache: dict | None):
"""Returns xpubs used to derive pubkeys using global xpub field from psbt"""
cosigners = []
for i, pubkey in enumerate(pubkeys):
@@ -338,7 +433,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
@@ -418,7 +515,7 @@ def verify_multisig_output(self, descriptor: Descriptor, change_num: int) -> boo
return is_owner
- def _fill_missing_fingerprints(self):
+ def _fill_missing_fingerprints(self, child_key_derivation_cache: dict):
"""
Fix for when fingerprint is missing (defaults to all zeros). Happens when the user
creates a new wallet in an external coordinator but only provides the xpub
@@ -431,11 +528,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 +543,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, 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,11 @@
import random
from binascii import a2b_base64
+from copy import deepcopy
+from unittest.mock import patch
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 +488,287 @@ 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: that each one actually
+ takes effect, and that none of them changes the result of a parse.
+ """
+ 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 assert_same_parse_result(self, parser_a: PSBTParser, parser_b: PSBTParser):
+ """
+ Asserts that two parses produced the same result, field by field so that a failure
+ names the exact field that differs.
+
+ The fill path writes recovered fingerprints back into the psbt, so the serialized
+ psbt is compared too, not just the parser's own attributes.
+ """
+ assert parser_a.policy == parser_b.policy
+ assert parser_a.input_amount == parser_b.input_amount
+ assert parser_a.spend_amount == parser_b.spend_amount
+ assert parser_a.change_amount == parser_b.change_amount
+ assert parser_a.fee_amount == parser_b.fee_amount
+ assert parser_a.num_inputs == parser_b.num_inputs
+ assert parser_a.destination_addresses == parser_b.destination_addresses
+ assert parser_a.destination_amounts == parser_b.destination_amounts
+ assert parser_a.change_data == parser_b.change_data
+ assert parser_a.op_return_data == parser_b.op_return_data
+ assert parser_a.psbt.serialize() == parser_b.psbt.serialize()
+
+
+ def cache_size_recorder(self, cache_sizes: list):
+ """
+ Returns a stand-in for _derive_with_cache that derives exactly as the real one
+ does, but appends the cache's size to cache_sizes on the way out of every call.
+
+ The cache is a local inside parse(), so intercepting the calls it gets handed to
+ is the only way to see how large it grew.
+ """
+ real_derive_with_cache = PSBTParser._derive_with_cache
+
+ def recorded(parent_key, derivation_path, cache=None):
+ derived_key = real_derive_with_cache(parent_key, derivation_path, cache)
+ cache_sizes.append(len(cache))
+ return derived_key
+
+ return recorded
+
+
+ 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_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)
+
+ # Two levels should have been added for each cosigner
+ assert len(cache) == 4
+
+ # The resulting derived child keys should be different
+ 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 it derives every level itself or reads them back out of the cache.
+ """
+ psbt = PSBT.parse(a2b_base64(PSBTTestData.MULTISIG_NATIVE_SEGWIT_1_INPUT))
+ inp = psbt.inputs[0]
+ pubkeys = list(inp.bip32_derivations.keys())
+
+ # No cache at all; every level is derived directly
+ uncached = PSBTParser._get_cosigners(pubkeys, inp.bip32_derivations, psbt.xpubs, None)
+
+ # An empty cache still has to derive every level, but now stores each one
+ child_key_derivation_cache = {}
+ populating_the_cache = PSBTParser._get_cosigners(pubkeys, inp.bip32_derivations, psbt.xpubs, child_key_derivation_cache)
+
+ # 3 cosigners x 2 levels each
+ assert len(child_key_derivation_cache) == 6
+
+ # The same call against the now-populated cache reads those levels back instead
+ # of deriving them. Each level sits below a different cosigner's xpub, so a cache
+ # that confused parents would return the wrong cosigner here.
+ reading_from_the_cache = PSBTParser._get_cosigners(pubkeys, inp.bip32_derivations, psbt.xpubs, child_key_derivation_cache)
+
+ assert populating_the_cache == uncached
+ assert reading_from_the_cache == uncached
+
+
+ def test_cache_does_not_change_parse_output(self):
+ """
+ The whole point of the cache is that it changes nothing at all. Parse the same
+ psbt twice — once normally, once with the cache discarded so that every derivation
+ falls through to embit's own HDKey.derive() — and require identical parser state
+ and identical resulting psbt bytes.
+
+ Single-sig and multisig each get a run because they reach the cache from different
+ starting points: single-sig traverses down from our own root, multisig down from
+ each cosigner's account xpub.
+ """
+ def build_psbt(input_base64: str, change_hex: str) -> PSBT:
+ # A fresh psbt for each parse: the base psbt plus its change output, twice.
+ psbt = PSBT.parse(a2b_base64(input_base64))
+ psbt.outputs.append(create_output(change_hex, 10_000))
+
+ # Add a duplicate output to ensure that the cache yields some hits; the second
+ # output will traverse the same levels the first one just cached.
+ psbt.outputs.append(create_output(change_hex, 10_000))
+ return psbt
+
+ def assert_cache_makes_no_difference(input_base64: str, change_hex: str):
+ # Store the real function before the patches below replace it. Each replacement
+ # still needs access to the real function to do the actual deriving.
+ real_derive_with_cache = PSBTParser._derive_with_cache
+
+ # This version of the replacement will derive exactly as the real cache-backed
+ # function does, but will also record the cache it was handed on each call.
+ caches_received = []
+ def recording_derive_with_cache(parent_key, derivation_path, cache=None):
+ caches_received.append(cache)
+ return real_derive_with_cache(parent_key, derivation_path, cache)
+
+ with patch.object(PSBTParser, "_derive_with_cache", staticmethod(recording_derive_with_cache)):
+ with_cache = PSBTParser(
+ build_psbt(input_base64, change_hex), self.seed, network=SettingsConstants.REGTEST)
+
+ # Sanity check: the cache was actually available during the parse
+ assert any(cache is not None for cache in caches_received)
+
+ # And then this version discards the cache, which sends the real function down
+ # its no-cache branch.
+ def cache_free_derive(parent_key, derivation_path, cache=None):
+ return real_derive_with_cache(parent_key, derivation_path)
+
+ with patch.object(PSBTParser, "_derive_with_cache", staticmethod(cache_free_derive)):
+ without_cache = PSBTParser(
+ build_psbt(input_base64, change_hex), self.seed, network=SettingsConstants.REGTEST)
+
+ # Regardless of whether or not the cache was available, the resulting parser
+ # state should be identical.
+ self.assert_same_parse_result(with_cache, without_cache)
+
+ assert_cache_makes_no_difference(PSBTTestData.SINGLE_SIG_NATIVE_SEGWIT_1_INPUT, PSBTTestData.SINGLE_SIG_NATIVE_SEGWIT_CHANGE)
+ assert_cache_makes_no_difference(PSBTTestData.MULTISIG_NATIVE_SEGWIT_1_INPUT, PSBTTestData.MULTISIG_NATIVE_SEGWIT_CHANGE)
+
+
+ def test_maxed_out_cache_does_not_change_parse_output(self):
+ """
+ There should be no effect on the parse output when the cache is maxed out.
+
+ Parse a multisig and a single-sig psbt with the cache free to grow, then parse
+ them again with the cap low enough that both hit the cap partway through. Verify
+ that we get the identical parser state each time.
+ """
+ multisig_case = (PSBTTestData.MULTISIG_NATIVE_SEGWIT_1_INPUT, PSBTTestData.MULTISIG_NATIVE_SEGWIT_CHANGE)
+ singlesig_case = (PSBTTestData.SINGLE_SIG_NATIVE_SEGWIT_1_INPUT, PSBTTestData.SINGLE_SIG_NATIVE_SEGWIT_CHANGE)
+
+ def build_psbt(case: tuple) -> PSBT:
+ # A fresh psbt for each parse: the case's base psbt plus its change output
+ input_base64, change_hex = case
+ psbt = PSBT.parse(a2b_base64(input_base64))
+ psbt.outputs.append(create_output(change_hex, 10_000))
+ return psbt
+
+ # Record how large the cache grew over the course of each parse
+ unconstrained_sizes = []
+ with patch.object(PSBTParser, "_derive_with_cache", staticmethod(self.cache_size_recorder(unconstrained_sizes))):
+ multisig_unconstrained = PSBTParser(build_psbt(multisig_case), self.seed, network=SettingsConstants.REGTEST)
+ singlesig_unconstrained = PSBTParser(build_psbt(singlesig_case), self.seed, network=SettingsConstants.REGTEST)
+
+ # Now constrain the cache enough that both psbts fill it partway through their
+ # parse.
+ cap = 3
+ capped_sizes = []
+ with patch.object(PSBTParser, "MAX_CACHED_DERIVATIONS", cap):
+ with patch.object(PSBTParser, "_derive_with_cache", staticmethod(self.cache_size_recorder(capped_sizes))):
+ multisig_capped = PSBTParser(build_psbt(multisig_case), self.seed, network=SettingsConstants.REGTEST)
+ singlesig_capped = PSBTParser(build_psbt(singlesig_case), self.seed, network=SettingsConstants.REGTEST)
+
+ self.assert_same_parse_result(multisig_unconstrained, multisig_capped)
+ self.assert_same_parse_result(singlesig_unconstrained, singlesig_capped)
+
+ # Sanity check: this test depends on the unconstrained cache actually being larger
+ # than the capped cache's max.
+ assert max(unconstrained_sizes) > cap
+ assert max(capped_sizes) == cap
+
+Why this scored 19/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.