"""
bal.core.will
=============
The "will": the set of time-locked inheritance transactions plus all the logic
to keep it coherent over time.
Two classes live here:
* :class:`Will` - a namespace of static methods operating on a *will*
dictionary (mapping ``txid -> WillItem``): building
the parent/child tree, anticipating locktimes,
detecting replaced/invalidated/confirmed entries,
validating that the will still matches the heirs and
will-executors, and building an "invalidation"
transaction.
* :class:`WillItem` - a single will transaction together with its status
flags, heirs, will-executor and fee.
Separation of concerns
-----------------------
The original ``WillItem`` carried a ``get_color()`` method returning hard-coded
hex colours for the GUI. That was pure presentation living inside the core
logic, so it has been **moved** to ``bal.gui.qt.theme.status_color(will_item)``.
The status flags themselves (the source of truth) stay here; only the mapping
"status -> colour" now lives in the GUI layer. No behaviour changed.
"""
import copy
from datetime import datetime, timezone
from electrum.i18n import _
from electrum.logging import Logger, get_logger
from electrum.transaction import (
PartialTransaction,
PartialTxInput,
PartialTxOutput,
Transaction,
TxOutpoint,
tx_from_any,
)
from electrum.util import (
UnrelatedTransactionException,
bfh,
)
from .heirs import WillExecutorFeeTooHighException
from .util import Util
from .willexecutors import Willexecutors
MIN_LOCKTIME = 1
MIN_BLOCK = 1
_logger = get_logger(__name__)
class Will:
@staticmethod
def get_children(will, willid):
out = []
for _id in will:
inputs = will[_id].tx.inputs()
for idi in range(0, len(inputs)):
_input = inputs[idi]
if _input.prevout.txid.hex() == willid:
out.append([_id, idi, _input.prevout.out_idx])
return out
# build a tree with parent transactions
@staticmethod
def add_willtree(will):
for willid in will:
will[willid].children = Will.get_children(will, willid)
for child in will[willid].children:
if not will[child[0]].father:
will[child[0]].father = willid
# return a list of will sorted by locktime
@staticmethod
def get_sorted_will(will):
return sorted(will.items(), key=lambda x: x[1]["tx"].locktime)
@staticmethod
def only_valid(will):
for k, v in will.items():
if v.get_status("VALID"):
yield k
@staticmethod
def needs_server_check(w):
"""Return True if ``w`` should be queried on its will-executor server
when the user presses Check (or on Electrum close).
A will is queried only when it is VALID, has a will-executor assigned,
was actually PUSHED (sent), and is not yet CHECKED. The ``PUSHED``
condition is essential: querying the server for a will that was *never*
sent would make the server (correctly) answer "I don't have this tx",
which ``WillItem.set_check_willexecutor`` then records as CHECK_FAIL.
A freshly signed-but-not-sent will would therefore turn red, even though
it is merely "signed, not sent" (which must stay blue / #2bc8ed, as in
the original BAL behaviour). Restricting the check to PUSHED wills
matches the original ``check()`` logic and avoids that false failure.
"""
return bool(
w.get_status("VALID")
and w.we
and w.get_status("PUSHED")
and not w.get_status("CHECKED")
)
@staticmethod
def search_equal_tx(will, tx, wid):
for w in will:
if w != wid and not tx.to_json() != will[w]["tx"].to_json():
if will[w]["tx"].txid() != tx.txid():
if Util.cmp_txs(will[w]["tx"], tx):
return will[w]["tx"]
return False
@staticmethod
def get_tx_from_any(x):
try:
a = str(x)
return tx_from_any(a)
except Exception as e:
raise e
return x
@staticmethod
def add_info_from_will(will, wid, wallet):
if isinstance(will[wid].tx, str):
will[wid].tx = Will.get_tx_from_any(will[wid].tx)
if wallet:
will[wid].tx.add_info_from_wallet(wallet)
for txin in will[wid].tx.inputs():
txid = txin.prevout.txid.hex()
if txid in will:
change = will[txid].tx.outputs()[txin.prevout.out_idx]
txin._trusted_value_sats = change.value
try:
txin.script_descriptor = change.script_descriptor
except Exception:
pass
txin.is_mine = True
txin._TxInput__address = change.address
txin._TxInput__scriptpubkey = change.scriptpubkey
txin._TxInput__value_sats = change.value
txin._trusted_value_sats = change.value
@staticmethod
def normalize_will(will, wallet=None, others_inputs=None):
others_input = others_inputs if others_inputs is not None else {}
to_delete = []
to_add = {}
# add info from wallet
willitems = {}
for wid in will:
Will.add_info_from_will(will, wid, wallet)
willitems[wid] = WillItem(will[wid])
will = willitems
errors = {}
for wid in will:
txid = will[wid].tx.txid()
if txid is None:
_logger.error("##########")
_logger.error(wid)
_logger.error(will[wid])
_logger.error(will[wid].tx.to_json())
_logger.error("txid is none")
will[wid].set_status("ERROR", True)
errors[wid] = will[wid]
continue
if txid != wid:
outputs = will[wid].tx.outputs()
ow = will[wid]
ow.normalize_locktime(others_input)
will[wid] = WillItem(ow.to_dict())
for i in range(0, len(outputs)):
Will.change_input(
will, wid, i, outputs[i], others_input, to_delete, to_add
)
to_delete.append(wid)
to_add[ow.tx.txid()] = ow.to_dict()
# for eid, err in errors.items():
# new_txid = err.tx.txid()
for k, w in to_add.items():
will[k] = w
for wid in to_delete:
if wid in will:
del will[wid]
@staticmethod
def new_input(txid, idx, change):
prevout = TxOutpoint(txid=bfh(txid), out_idx=idx)
inp = PartialTxInput(prevout=prevout)
inp._trusted_value_sats = change.value
inp.is_mine = True
inp._TxInput__address = change.address
inp._TxInput__scriptpubkey = change.scriptpubkey
inp._TxInput__value_sats = change.value
return inp
@staticmethod
def check_anticipate(ow: "WillItem", nw: "WillItem"):
anticipate = Util.anticipate_locktime(ow.tx.locktime, days=1)
if int(nw.tx.locktime) >= int(anticipate):
if Util.cmp_heirs_by_values(
ow.heirs, nw.heirs, [0, 1], exclude_willexecutors=True
):
if nw.we and ow.we:
if ow.we["url"] == nw.we["url"]:
if int(ow.we["base_fee"]) > int(nw.we["base_fee"]):
return anticipate
elif int(ow.tx_fees) != int(nw.tx_fees):
return anticipate
else:
if nw.we == ow.we:
if not Util.cmp_heirs_by_values(ow.heirs, nw.heirs, [0, 3]):
return anticipate
else:
return ow.tx.locktime
else:
return ow.tx.locktime
else:
return anticipate
return 4294967295 + 1
@staticmethod
def change_input(will, otxid, idx, change, others_inputs, to_delete, to_append):
ow = will[otxid]
ntxid = ow.tx.txid()
if otxid != ntxid:
for wid in will:
w = will[wid]
inputs = w.tx.inputs()
outputs = w.tx.outputs()
found = False
old_txid = w.tx.txid()
# ntx = None
for i in range(0, len(inputs)):
if (
inputs[i].prevout.txid.hex() == otxid
and inputs[i].prevout.out_idx == idx
):
if isinstance(w.tx, Transaction):
will[wid].tx = PartialTransaction.from_tx(w.tx)
will[wid].tx.set_rbf(True)
will[wid].tx._inputs[i] = Will.new_input(wid, idx, change)
found = True
if found:
pass
new_txid = will[wid].tx.txid()
if old_txid != new_txid:
to_delete.append(old_txid)
to_append[new_txid] = will[wid]
outputs = will[wid].tx.outputs()
for i in range(0, len(outputs)):
Will.change_input(
will,
wid,
i,
outputs[i],
others_inputs,
to_delete,
to_append,
)
@staticmethod
def get_all_inputs(will, only_valid=False):
all_inputs = {}
for w, wi in will.items():
if not only_valid or wi.get_status("VALID"):
inputs = wi.tx.inputs()
for i in inputs:
prevout_str = i.prevout.to_str()
inp = [w, will[w], i]
if prevout_str not in all_inputs:
all_inputs[prevout_str] = [inp]
else:
all_inputs[prevout_str].append(inp)
return all_inputs
@staticmethod
def get_all_inputs_min_locktime(all_inputs):
all_inputs_min_locktime = {}
for i, values in all_inputs.items():
min_locktime = min(values, key=lambda x: x[1].tx.locktime)[1].tx.locktime
for w in values:
if w[1].tx.locktime == min_locktime:
if i not in all_inputs_min_locktime:
all_inputs_min_locktime[i] = [w]
else:
all_inputs_min_locktime[i].append(w)
return all_inputs_min_locktime
@staticmethod
def search_anticipate_rec(will, old_inputs):
redo = False
to_delete = []
to_append = {}
new_inputs = Will.get_all_inputs(will, only_valid=True)
for nid, nwi in will.items():
if nwi.search_anticipate(new_inputs):
if nid != nwi.tx.txid():
redo = True
to_delete.append(nid)
to_append[nwi.tx.txid()] = nwi
outputs = nwi.tx.outputs()
for i in range(0, len(outputs)):
Will.change_input(
will, nid, i, outputs[i], new_inputs, to_delete, to_append
)
if nwi.search_anticipate(old_inputs):
if nid != nwi.tx.txid():
redo = True
to_delete.append(nid)
to_append[nwi.tx.txid()] = nwi
outputs = nwi.tx.outputs()
for i in range(0, len(outputs)):
Will.change_input(
will, nid, i, outputs[i], new_inputs, to_delete, to_append
)
for w in to_delete:
try:
del will[w]
except Exception:
pass
for k, w in to_append.items():
will[k] = w
if redo:
Will.search_anticipate_rec(will, old_inputs)
@staticmethod
def _same_heirs(old_heirs, new_heirs):
"""Return True if two heir maps describe the SAME inheritance.
Used by update_will (Option A) to decide whether a rebuilt transaction
that kept the same txid can safely reuse the old (possibly already
signed) WillItem, or whether the heirs changed and the item must be
rebuilt as unsigned.
Two heir maps are considered equal when they have exactly the same heir
names (keys) and, for each heir, the same destination ADDRESS, the same
requested AMOUNT and the same LOCKTIME. Internal will-executor
pseudo-heirs (keys starting with the reserved ``w!ll3x3c"`` prefix) are
ignored, exactly as in check_willexecutors_and_heirs, because they are
bookkeeping entries and not real heirs.
Args:
old_heirs: heirs dict stored in the old (existing) WillItem.
new_heirs: heirs dict of the freshly rebuilt WillItem.
Returns:
bool: True if the real heirs are identical, False otherwise.
"""
def _real_heirs(heirs):
# Keep only the real heirs and only the fields that define the
# inheritance (address/amount/locktime), so cosmetic or derived
# fields can never trigger a spurious "heirs changed" rebuild.
out = {}
for name, entry in (heirs or {}).items():
if str(name)[:9] == 'w!ll3x3c"':
continue
# Heir entry layout (see heirs.py): [0]=address, [1]=amount,
# [2]=locktime. We compare exactly the same fields that
# check_willexecutors_and_heirs uses (their[0], their[1],
# their[2]); index literals are used here to avoid importing the
# heirs module (which would create a circular import).
out[name] = (entry[0], entry[1], entry[2])
return out
return _real_heirs(old_heirs) == _real_heirs(new_heirs)
@staticmethod
def update_will(old_will, new_will):
all_old_inputs = Will.get_all_inputs(old_will, only_valid=True)
# all_inputs_min_locktime = Will.get_all_inputs_min_locktime(all_old_inputs)
# all_new_inputs = Will.get_all_inputs(new_will)
# check if the new input is already spent by other transaction
# if it is use the same locktime, or anticipate.
Will.search_anticipate_rec(new_will, all_old_inputs)
other_inputs = Will.get_all_inputs(old_will, {})
try:
Will.normalize_will(new_will, others_inputs=other_inputs)
except Exception as e:
raise e
for oid in Will.only_valid(old_will):
if oid in new_will:
new_heirs = new_will[oid].heirs
new_we = new_will[oid].we
# OPTION A (heir-change full rebuild, user-approved):
#
# Historically, whenever a rebuilt transaction kept the SAME
# txid as an old one, we REUSED the old WillItem object (which
# may already be signed/COMPLETE/PUSHED) and only copied the new
# heirs/will-executor onto it. That silently preserved the
# "already signed" status even when the HEIRS had actually
# changed (e.g. an heir was deleted, so amounts must be
# recomputed and the whole wallet re-swept). The downstream
# have_to_sign check then saw the item as COMPLETE and reported
# "Nothing to do", so the new will was never signed/broadcast
# (bugs E/F/K).
#
# We now reuse the old item ONLY when the heirs are IDENTICAL.
# If the heir set/values changed, we keep the freshly built
# item (status "New", not COMPLETE) so it is correctly detected
# as needing a new signature and broadcast. The will-executor is
# still refreshed in both cases.
if Will._same_heirs(old_will[oid].heirs, new_heirs):
new_will[oid] = old_will[oid]
new_will[oid].heirs = new_heirs
new_will[oid].we = new_we
else:
# Heirs changed: keep the new (unsigned) item but make sure
# it carries the up-to-date will-executor.
new_will[oid].we = new_we
continue
else:
continue
@staticmethod
def get_higher_input_for_tx(will):
out = {}
for wid in will:
wtx = will[wid].tx
found = False
inp = None
for inp in wtx.inputs():
if inp.prevout.txid.hex() in will:
found = True
break
if not found and inp is not None:
out[inp.prevout.to_str()] = inp
return out
@staticmethod
def invalidate_will(will, wallet, fees_per_byte, history_label=None,
will_locktime=None):
_logger.debug("invalidate tx in will module")
will_only_valid = Will.only_valid_list(will)
inputs = Will.get_all_inputs(will_only_valid)
if history_label is not None and will_locktime is not None:
utxos = Util.get_available_utxos(wallet, history_label, will_locktime)
else:
utxos = wallet.get_utxos()
filtered_inputs = []
prevout_to_spend = []
current_height = Util.get_current_height(wallet.network)
for prevout_str, ws in inputs.items():
for w in ws:
if w[0] not in filtered_inputs:
filtered_inputs.append(w[0])
if prevout_str not in prevout_to_spend:
prevout_to_spend.append(prevout_str)
balance = 0
utxo_to_spend = []
for utxo in utxos:
if utxo.is_coinbase_output() and utxo.block_height < current_height+100:
_logger.debug("is not mature coinbase output")
continue
utxo_str = utxo.prevout.to_str()
if utxo_str in prevout_to_spend:
balance += inputs[utxo_str][0][2].value_sats()
utxo_to_spend.append(utxo)
_logger.debug("utxo to spend: {}".format(utxo_to_spend))
if len(utxo_to_spend) > 0:
change_addresses = wallet.get_change_addresses_for_new_transaction()
out = PartialTxOutput.from_address_and_value(change_addresses[0], balance)
out.is_change = True
locktime = current_height
tx = PartialTransaction.from_io(
utxo_to_spend, [out], locktime=locktime, version=2
)
tx.set_rbf(True)
fee = tx.estimated_size() * fees_per_byte
if balance - fee > 0:
out = PartialTxOutput.from_address_and_value(
change_addresses[0], balance - fee
)
tx = PartialTransaction.from_io(
utxo_to_spend, [out], locktime=locktime, version=2
)
tx.set_rbf(True)
_logger.debug(f"invalidation tx: {tx}")
return tx
else:
_logger.debug(f"balance({balance}) - fee({fee}) <=0")
pass
else:
_logger.debug("len utxo_to_spend <=0")
pass
@staticmethod
def is_new(will):
for _wid, w in will.items():
if w.get_status("VALID") and not w.get_status("COMPLETE"):
return True
@staticmethod
def search_rai(all_inputs, all_utxos, will, wallet):
# will_only_valid = Will.only_valid_or_replaced_list(will)
for inp, ws in all_inputs.items():
inutxo = Util.in_utxo(inp, all_utxos)
for w in ws:
wi = w[1]
if (
wi.get_status("VALID")
or wi.get_status("CONFIRMED")
or wi.get_status("MEMPOOL")
):
prevout_id = w[2].prevout.txid.hex()
if not inutxo:
if prevout_id in will:
wo = will[prevout_id]
if wo.get_status("REPLACED"):
wi.set_status("REPLACED", True)
if wo.get_status("INVALIDATED"):
wi.set_status("INVALIDATED", True)
else:
# The funding outpoint is not part of the will tree:
# decide from whether a broadcast transaction really
# spends it (a wallet-local history copy of the same
# will tx must neither turn the item CONFIRMED nor
# INVALIDATED - it is just a persistence artifact).
stored = None
if wallet and getattr(wallet, "db", None):
try:
stored = wallet.db.get_transaction(wi._id)
except Exception:
stored = None
spender_height = Will._funding_spender_height(wallet, inp)
if spender_height is None:
if stored:
continue
wi.set_status("INVALIDATED", True)
elif spender_height == 0:
wi.set_status("MEMPOOL", True)
else:
wi.set_status("CONFIRMED", True)
for child in wi.search(all_inputs):
if child.tx.locktime < wi.tx.locktime:
_logger.debug("a child was found")
wi.set_status("REPLACED", True)
else:
pass
@staticmethod
def utxos_strs(utxos):
return [Util.utxo_to_str(u) for u in utxos]
@staticmethod
def set_invalidate(wid, will=None):
will = will if will is not None else {}
will[wid].set_status("INVALIDATED", True)
if will[wid].children:
for c in will[wid].children.items():
Will.set_invalidate(c[0], will)
@staticmethod
def check_tx_height(tx, wallet):
info = wallet.get_tx_info(tx)
return info.tx_mined_status.height()
# check if transactions are stil valid tecnically valid
@staticmethod
def check_invalidated(willtree, utxos_list, wallet):
for wid, w in willtree.items():
if (
not w.father
or willtree[w.father].get_status("CONFIRMED")
or willtree[w.father].get_status("MEMPOOL")
):
for inp in w.tx.inputs():
inp_str = Util.utxo_to_str(inp)
if inp_str not in utxos_list:
if not wallet or not getattr(wallet, "adb", None):
continue
height = Will.check_tx_height(w.tx, wallet)
if height < 0:
# The will tx itself is not on-chain. A missing
# funding UTXO is only a real problem when a
# broadcast transaction actually spends it; a
# wallet-local (history) copy of the same will tx
# marks the funding spent locally and must not
# invalidate the will.
if Will._funding_really_spent(wallet, inp_str):
Will.set_invalidate(wid, willtree)
elif height == 0:
w.set_status("MEMPOOL", True)
else:
w.set_status("CONFIRMED", True)
# def reflect_to_children(treeitem):
# if not treeitem.get_status("VALID"):
# _logger.debug(f"{tree:item._id} status not valid looking for children")
# for child in treeitem.children:
# wc = willtree[child]
# if wc.get_status("VALID"):
# if treeitem.get_status("INVALIDATED"):
# wc.set_status("INVALIDATED", True)
# if treeitem.get_status("REPLACED"):
# wc.set_status("REPLACED", True)
# if wc.children:
# Will.reflect_to_children(wc)
@staticmethod
def check_amounts(heirs, willexecutors, all_utxos, timestamp_to_check, dust,
max_fee=None):
fixed_heirs, fixed_amount, perc_heirs, perc_amount, fixed_amount_with_dust = (
heirs.fixed_percent_lists_amount(timestamp_to_check, dust, reverse=True)
)
wallet_balance = 0
for utxo in all_utxos:
wallet_balance += utxo.value_sats()
if fixed_amount >= wallet_balance:
raise FixedAmountException(
f"Fixed amount({fixed_amount}) >= {wallet_balance}"
)
if perc_amount != 100:
raise PercAmountException(f"Perc amount({perc_amount}) =! 100%")
for url, wex in willexecutors.items():
if Willexecutors.is_selected(wex) and Willexecutors.is_valid(wex, max_fee=max_fee, dust=dust):
if max_fee is not None and int(wex["base_fee"]) > max_fee:
raise WillExecutorFeeTooHighException(wex, max_fee)
temp_balance = wallet_balance - int(wex["base_fee"])
if fixed_amount >= temp_balance:
raise FixedAmountException(
f"Willexecutor{url} excess base fee({wex['base_fee']}), {fixed_amount} >={temp_balance}"
)
@staticmethod
def _funding_really_spent(wallet, inp_str):
"""True when a broadcast transaction really spends the ``txid:n`` outpoint.
``wallet.adb.get_spender`` discards wallet-local spenders (the stored
will tx from the local history) and future transactions, so this is True
only when the funding was consumed by a real on-chain/mempool tx.
"""
if not wallet or not getattr(wallet, "adb", None):
return False
try:
return wallet.adb.get_spender(inp_str) is not None
except Exception as e:
_logger.error(f"get_spender failed for {inp_str}: {e}")
return False
@staticmethod
def _funding_spender_height(wallet, inp_str):
"""Mined height of the broadcast tx spending ``inp_str``, or None.
Returns ``None`` when no broadcast transaction spends the outpoint (a
wallet-local history spender or a future tx are ignored by
``adb.get_spender``). The height is 0 for a mempool spender and positive
for a confirmed one.
"""
if not wallet or not getattr(wallet, "adb", None):
return None
try:
spender = wallet.adb.get_spender(inp_str)
except Exception as e:
_logger.error(f"get_spender failed for {inp_str}: {e}")
return None
if spender is None:
return None
try:
return int(wallet.adb.get_tx_height(spender).height())
except Exception as e:
_logger.error(f"get_tx_height failed for {spender}: {e}")
return 0
@staticmethod
def _absorb_history_signatures(will, wallet):
"""Merge signatures from the wallet's stored local copy of each will tx.
An incomplete will transaction saved into the local history (see
``save_valid_transactions_to_history``) may later accumulate signatures
(e.g. after a manual merge from a more complete copy). The in-memory
will item would otherwise miss those signatures on the next check. For
every item whose stored wallet copy is the same partial transaction the
signatures are merged into the in-memory one and, if it becomes fully
signed, the item is marked COMPLETE.
This method must never raise: history absorption is a convenience on top
of the will check, so any failure is logged and ignored.
"""
if not wallet or not getattr(wallet, "db", None):
return
for wi in will.values():
try:
if (
wi.tx is None
or not isinstance(wi.tx, PartialTransaction)
or wi.tx.is_complete()
):
continue
stored = wallet.db.get_transaction(wi._id)
if (
not isinstance(stored, Transaction)
or stored.txid() != wi.tx.txid()
):
continue
wi.tx.combine_with_other_psbt(stored)
if wi.tx.is_complete():
wi.set_status("COMPLETE", True)
except Exception as e:
_logger.error(f"absorb history signatures failed for item {wi._id}: {e}")
@staticmethod
def check_will(will, all_utxos, wallet, timestamp_to_check):
"""Validate a will against the current wallet state.
Locktimes are always UNIX timestamps (block-height locktimes are no
longer supported by this plugin), so expiry is decided purely by
comparing each transaction's locktime against ``timestamp_to_check``.
Args:
will: The will dictionary (WillItem entries keyed by txid).
all_utxos: The list of UTXOs currently available in the wallet.
wallet: The Electrum wallet object.
timestamp_to_check: The reference UNIX timestamp (usually "now")
used to decide whether any transaction has expired.
"""
Will._absorb_history_signatures(will, wallet)
Will.add_willtree(will)
utxos_list = Will.utxos_strs(all_utxos)
Will.check_invalidated(will, utxos_list, wallet)
all_inputs = Will.get_all_inputs(will, only_valid=True)
all_inputs_min_locktime = Will.get_all_inputs_min_locktime(all_inputs)
Will.check_will_expired(all_inputs_min_locktime, timestamp_to_check)
all_inputs = Will.get_all_inputs(will, only_valid=True)
Will.search_rai(all_inputs, all_utxos, will, wallet)
Will.check_signatures(will, wallet)
@staticmethod
def save_valid_transactions_to_history(will, wallet, history_label):
"""Keep the wallet's LOCAL history in sync with the current will state.
Called after the will has been built/signed/checked (see the
SAVE_HISTORY / HISTORY_LABEL settings). A will transaction belongs in the
local history while it is still "New" (not yet fully signed - i.e. an
incomplete partial transaction), and must be removed once it becomes
"Complete" (fully signed), because at that point it is ready to be
broadcast and will appear in the history on its own.
For every will item that is valid and whose transaction has a txid it:
1. decodes the label template, replacing "{willexecutor}" with the
will-executor URL of the item,
2. if the transaction is NOT complete, stores it via
``wallet.adb.add_transaction`` (merging signatures when an
already-stored partial transaction is upgraded by a more complete
one) and tags it with the decoded label,
3. if the transaction IS complete, does not store it: its matching
local-history entry is removed by the cleanup below.
Finally it deletes every wallet-local transaction whose label exactly
matches the decoded label of a current valid item but that is no longer
among the just-saved transactions, so fully-signed, rebuilt or replaced
wills do not pile up stale entries.
This method must never raise: history persistence is a convenience on
top of the will check, so any failure is logged and ignored.
Args:
will: The will dictionary (WillItem entries keyed by txid).
wallet: The Electrum wallet object (may be falsy for offline
checks, in which case this is a no-op).
history_label: The label template to apply (may contain
"{willexecutor}").
"""
if not wallet or not getattr(wallet, "adb", None):
return
saved_txids = []
try:
current_labels = {
history_label.replace(
"{willexecutor}", (wi.we or {}).get("url", "")
)
for wi in will.values()
if wi.get_status("VALID")
and wi.tx is not None
and wi.tx.txid() is not None
}
for wi in will.values():
if not wi.get_status("VALID"):
continue
if wi.tx is None or wi.tx.txid() is None:
continue
# Fully-signed (complete) transactions must NOT be saved: they
# are removed from the local history so the list does not show a
# placeholder for a transaction that will appear on its own once
# broadcast/confirmed. Only the not-yet-complete "New" items are
# stored. Note that fully-segwit partial txs have a txid even
# when incomplete, so the txid() check alone is not enough.
if wi.tx.is_complete():
continue
try:
txid = wi.tx.txid()
label = history_label.replace(
"{willexecutor}", (wi.we or {}).get("url", "")
)
Will._add_transaction_to_history(wallet, wi.tx, txid)
try:
wallet.set_label(txid, label)
except Exception as e:
_logger.error(f"set_label failed for {txid}: {e}")
saved_txids.append(txid)
except Exception as e:
_logger.error(f"save to history failed for item {wi._id}: {e}")
# Delete stale wallet-local txs whose label matches a current valid
# item but that are no longer among the saved ones. This removes
# entries for fully-signed (complete) items and for rebuilt/replaced
# wills with the same executor.
for txid, label in Will._wallet_labels(wallet):
if txid in saved_txids:
continue
if label not in current_labels:
continue
try:
wallet.adb.remove_transaction(txid)
try:
wallet.set_label(txid, None)
except Exception:
pass
except Exception as e:
_logger.error(f"remove from history failed for {txid}: {e}")
try:
wallet.save_db()
except Exception as e:
_logger.error(f"save_db failed after history update: {e}")
except Exception as e:
_logger.error(f"save_valid_transactions_to_history failed: {e}")
@staticmethod
def _add_transaction_to_history(wallet, tx, txid):
"""Store *tx* into the wallet's local history via ``adb``.
If a partial transaction with the same txid is already stored and *tx*
carries additional signatures, the signatures are merged into the stored
one before saving. ``allow_unrelated`` is retried as a fallback so that
self-created txs (which are not yet part of the wallet's UTXO set) are
still accepted.
"""
adb = wallet.adb
existing = None
try:
existing = wallet.db.get_transaction(txid)
except Exception:
existing = None
try:
if (
isinstance(existing, PartialTransaction)
and not existing.is_complete()
and isinstance(tx, PartialTransaction)
):
existing.combine_with_other_psbt(tx)
adb.add_transaction(existing)
else:
try:
adb.add_transaction(tx)
except UnrelatedTransactionException:
adb.add_transaction(tx, allow_unrelated=True)
except Exception as e:
raise RuntimeError(f"add_transaction failed for {txid}: {e}") from e
@staticmethod
def _wallet_labels(wallet):
"""Return the wallet's ``(txid, label)`` pairs in a defensive way."""
try:
get_all_labels = wallet.get_all_labels
except AttributeError:
return []
try:
return list(get_all_labels().items())
except Exception as e:
_logger.error(f"get_all_labels failed: {e}")
return []
@staticmethod
def check_signatures(will, wallet=None):
"""Refresh the per-item signature counts and the PARTIALLY_SIGNED status.
The signature counts are derived from the transaction itself via
Electrum's ``signature_count()``, which needs a script descriptor on
each input (attached from the wallet when available). Items that already
carry their own descriptors (e.g. imported/merged partial transactions)
are counted even without a wallet.
An item with at least one signature present but fewer than required is
marked PARTIALLY_SIGNED. Items that are already signed (COMPLETE) or
whose transaction is complete always clear the flag.
"""
for wi in will.values():
try:
if wi.get_status("COMPLETE") or wi.tx is None or wi.tx.is_complete():
wi.set_status("PARTIALLY_SIGNED", False)
continue
if wallet:
wi.tx.add_info_from_wallet(wallet)
if not hasattr(wi.tx, "signature_count"):
continue
have, required = wi.tx.signature_count()
wi.sigs_have = int(have)
wi.sigs_required = int(required)
if required > 1 and 0 < have < required:
wi.set_status("PARTIALLY_SIGNED", True)
else:
wi.set_status("PARTIALLY_SIGNED", False)
except Exception as e:
_logger.error(f"check_signatures failed for item {wi._id}: {e}")
@staticmethod
def get_min_locktime(will,default_value=None):
return min((v.tx.locktime for v in will.values() if v.get_status('VALID')), default=default_value)
@staticmethod
def is_will_valid(
will,
timestamp_to_check,
tx_fees,
all_utxos,
heirs=None,
willexecutors=None,
self_willexecutor=False,
wallet=False,
callback_not_valid_tx=None,
):
"""Check whether the whole will is valid at the given timestamp.
Locktimes are always UNIX timestamps, so the validity check only needs
a single reference timestamp (no block height).
Args:
will: The will dictionary (WillItem entries keyed by txid).
timestamp_to_check: Reference UNIX timestamp (usually "now").
tx_fees: Fee rate used for the dust/coverage check.
all_utxos: The list of UTXOs currently available in the wallet.
heirs: Optional heirs dictionary.
willexecutors: Optional will-executors dictionary.
self_willexecutor: Whether the user acts as their own executor.
wallet: The Electrum wallet object.
callback_not_valid_tx: Optional callback invoked for invalid txs.
Returns:
True if the will is valid; raises an exception otherwise.
"""
heirs = heirs if heirs is not None else {}
willexecutors= willexecutors if willexecutors is not None else {}
Will.check_will(will, all_utxos, wallet, timestamp_to_check)
if heirs:
if not Will.check_willexecutors_and_heirs(
will,
heirs,
willexecutors,
self_willexecutor,
timestamp_to_check,
tx_fees,
):
raise NotCompleteWillException()
all_inputs = Will.get_all_inputs(will, only_valid=True)
_logger.info("check all utxo in wallet are spent")
if all_inputs:
for utxo in all_utxos:
if utxo.value_sats() > 68 * tx_fees:
if not Util.in_utxo(utxo, all_inputs.keys()):
_logger.info("utxo is not spent", utxo.to_json())
_logger.debug(all_inputs.keys())
raise NotCompleteWillException(
"Some utxo in the wallet is not included"
)
_logger.info("will ok")
return True
@staticmethod
def _short_will_id(will_id):
"""Return a human-friendly, shortened form of a will id (hash).
Will ids are long hex strings (e.g. a 64-char txid) that are hard to
read in a message box. This keeps only the first and last 8 characters
joined by an ellipsis (e.g. ``9f1b0a75…fed9ae1b``). Short ids are left
untouched.
Args:
will_id: The will identifier (hash) to shorten; coerced to ``str``.
Returns:
str: The shortened id, or the original string if it is short.
"""
text = str(will_id)
# Only shorten when there is something to gain (>= 20 chars), otherwise
# the ellipsis form would not actually be shorter or clearer.
if len(text) <= 20:
return text
return f"{text[:8]}\u2026{text[-8:]}"
@staticmethod
def _format_locktime(locktime):
"""Format a UNIX locktime timestamp as a readable UTC date string.
Args:
locktime: UNIX timestamp (seconds) to format.
Returns:
str: A string like ``2026-06-22 11:00 UTC``. If formatting fails
for any reason, the raw timestamp is returned as a fallback so the
caller always has something to show.
"""
try:
dt = datetime.fromtimestamp(int(locktime), tz=timezone.utc)
return dt.strftime("%Y-%m-%d %H:%M UTC")
except Exception:
# Never let date formatting break the (already exceptional) flow.
return str(locktime)
@staticmethod
def check_will_expired(all_inputs_min_locktime, timestamp_to_check):
"""Raise WillExpiredException if any valid transaction has expired.
Locktimes are always UNIX timestamps, so a transaction is expired when
its locktime is in the past relative to ``timestamp_to_check``.
When a will is expired the message is written to be reassuring rather
than alarming: being past the locktime is an EXPECTED situation that the
plugin handles by invalidating the old will and re-signing it. The
message therefore uses a shortened will id and a human-readable UTC date
instead of raw values.
Args:
all_inputs_min_locktime: Mapping prevout -> will-item info, used to
find the minimum locktime per input.
timestamp_to_check: Reference UNIX timestamp (usually "now").
"""
_logger.info("check if some transaction is expired")
for _inputs, wid in all_inputs_min_locktime.items():
for w in wid:
if w[1].get_status("VALID"):
locktime = int(wid[0][1].tx.locktime)
# Locktimes are always timestamps: expired when in the past.
if locktime < int(timestamp_to_check):
# Build a clear, non-technical message: short id +
# readable date, and explain what will happen next.
short_id = Will._short_will_id(wid[0][0])
when = Will._format_locktime(locktime)
# The message is rendered as HTML by msg_warning(),
# so a
tag forces a line break: the second
# sentence goes on its own line to avoid an overly
# long single line in the wizard.
raise WillExpiredException(
"Will expired (id {id}, locktime {when}) \u2014
"
"too late to anticipate, the will will be "
"invalidated and re-signed.".format(
id=short_id, when=when
)
)
else:
_logger.debug(
f"Will Not Expired {wid[0][0]}: "
f"{Will._format_locktime(locktime)} > "
f"{Will._format_locktime(timestamp_to_check)}"
)
# def check_all_input_spent_are_in_wallet():
# _logger.info("check all input spent are in wallet or valid txs")
# for inp, ws in all_inputs.items():
# if not Util.in_utxo(inp, all_utxos):
# for w in ws:
# if w[1].get_status("VALID"):
# prevout_id = w[2].prevout.txid.hex()
# parentwill = will.get(prevout_id, False)
# if not parentwill or not parentwill.get_status("VALID"):
# w[1].set_status("INVALIDATED", True)
@staticmethod
def only_valid_list(will):
out = {}
for wid, w in will.items():
if w.get_status("VALID"):
out[wid] = w
return out
@staticmethod
def only_valid_or_replaced_list(will):
out = []
for wid, w in will.items():
wi = w
if wi.get_status("VALID") or wi.get_status("REPLACED"):
out.append(wid)
return out
@staticmethod
def check_willexecutors_and_heirs(
will, heirs, willexecutors, self_willexecutor, check_date, tx_fees
):
_logger.debug("check willexecutors heirs")
no_willexecutor = 0
willexecutors_found = {}
heirs_found = {}
will_only_valid = Will.only_valid_list(will)
if len(will_only_valid) < 1:
return False
for wid in Will.only_valid_list(will):
w = will[wid]
if w.tx_fees != tx_fees:
raise TxFeesChangedException(f"{tx_fees}: {w.tx_fees}")
for wheir in w.heirs:
if not 'w!ll3x3c"' == wheir[:9]:
their = will[wid].heirs[wheir]
if heir := heirs.get(wheir, None):
if heir[0] == their[0] and heir[1] == their[1]:
# The requested (possibly new) locktime for this heir.
new_locktime = Util.parse_locktime_string(heir[2])
# IMPORTANT: compare against the locktime that is
# actually frozen inside the already-signed Bitcoin
# transaction (w.tx.locktime), NOT against their[2].
# their[2] is the heir entry stored in the will item,
# which is updated in memory together with the new
# heirs dict when the user postpones, so it would
# always equal new_locktime and the postpone would go
# undetected. w.tx.locktime is immutable once signed
# and is exactly what the will-executors hold.
tx_locktime = int(w.tx.locktime)
if new_locktime == tx_locktime:
# Unchanged: this heir is still coherent.
count = heirs_found.get(wheir, 0)
heirs_found[wheir] = count + 1
elif new_locktime > tx_locktime and (
w.get_status("COMPLETE") or w.get_status("PUSHED")
):
# POSTPONE of an already signed/sent will: the
# old pre-signed tx must be invalidated on-chain
# first, otherwise a will-executor could
# broadcast the earlier-locktime tx and execute
# the inheritance too early.
raise WillPostponedException(
f"{wheir}: locktime postponed "
f"{tx_locktime}->{new_locktime} "
f"on a signed/sent will"
)
# ANTICIPATE (new_locktime < tx_locktime): the user
# manually moved the delivery date EARLIER.
# * If the new date is still in the FUTURE, this is
# a plain ANTICIPATE: it falls through here and is
# rebuilt via HeirNotFoundException (no on-chain
# fee). It must NEVER invalidate on-chain, even if
# the will was already signed/sent (A3, owner
# decision D2 = A1).
# * If the new date is in the PAST (relative to the
# check date) the will is genuinely expired and
# check_will_expired -> WillExpiredException handles
# it (on-chain invalidation). That is a different
# situation from "anticipate" and is intentionally
# kept.
#
# new_locktime > tx_locktime on a will that was never
# signed/sent also falls through here -> a plain rebuild
# via HeirNotFoundException (no on-chain fee needed).
else:
# The will still carries this heir, but the heir is no
# longer present in the current heirs set: the user
# removed it. This must trigger a rebuild exactly like
# "heir added" does, otherwise the removed heir would
# silently stay in the inheritance transaction. Raising
# HeirNotFoundException reuses the same rebuild path used
# by the Check button and by on_close (Electrum quit).
_logger.debug(
f"heir removed, transaction is not valid:"
f"{wheir} {wid}, {w}"
)
raise HeirNotFoundException(wheir)
if willexecutor := w.we:
count = willexecutors_found.get(willexecutor["url"], 0)
if Util.cmp_willexecutor(
willexecutor, willexecutors.get(willexecutor["url"], None)
):
willexecutors_found[willexecutor["url"]] = count + 1
else:
no_willexecutor += 1
count_heirs = 0
for h in heirs:
if Util.parse_locktime_string(heirs[h][2]) >= check_date:
count_heirs += 1
if h not in heirs_found:
_logger.debug(f"heir: {h} not found")
raise HeirNotFoundException(h)
if not count_heirs:
raise NoHeirsException("there are not valid heirs")
if self_willexecutor and no_willexecutor == 0:
raise NoWillExecutorNotPresent("Backup tx")
for url, we in willexecutors.items():
if Willexecutors.is_selected(we) and Willexecutors.is_valid(we):
if url not in willexecutors_found:
_logger.debug(f"will-executor: {url} not fount")
raise WillExecutorNotPresent(url)
_logger.info("will is coherent with heirs and will-executors")
return True
class WillItem(Logger):
# Default status flags for an inheritance transaction.
# Each entry maps an internal status key to [human-readable label, default
# boolean value].
#
# A2 changes:
# * "PENDING" was renamed to "MEMPOOL" (the transaction has been seen in
# the Electrum mempool). Old saved wills that still carry the legacy
# "PENDING" key are migrated to "MEMPOOL" in __init__ (see below), so
# nothing is lost.
# * "UPDATED" was added: the transaction was spendable AND valid, and a new
# transaction replaces it while keeping the SAME locktime and SAME heirs.
# UPDATED keeps the VALID flag (see set_status).
STATUS_DEFAULT = {
"ANTICIPATED": ["Anticipated", False],
"BROADCASTED": ["Broadcasted", False],
"CHECKED": ["Checked", False],
"CHECK_FAIL": ["Check Failed", False],
"COMPLETE": ["Signed", False],
"CONFIRMED": ["Confirmed", False],
"ERROR": ["Error", False],
"EXPIRED": ["Expired", False],
"EXPORTED": ["Exported", False],
"IMPORTED": ["Imported", False],
"INVALIDATED": ["Invalidated", False],
"MEMPOOL": ["Mempool", False],
"PUSH_FAIL": ["Push failed", False],
"PUSHED": ["Pushed", False],
"PARTIALLY_SIGNED": ["Partially Signed", False],
"REPLACED": ["Replaced", False],
"RESTORED": ["Restored", False],
"UPDATED": ["Updated", False],
"VALID": ["Valid", True],
}
def set_status(self, status, value=True):
"""Set a status flag and apply the related side effects.
Some statuses imply that other statuses must change. The rules below
match the inheritance state machine:
VALID handling:
* INVALIDATED, REPLACED, CONFIRMED, MEMPOOL -> clear VALID
(the transaction can no longer be delivered as a valid will tx).
* ANTICIPATED -> KEEPS VALID. Anticipating only moves the locktime
earlier by 1 day; the transaction stays valid (it is NOT in the
"clear VALID" list on purpose).
* UPDATED -> KEEPS VALID. The transaction is replaced by a new one
that keeps the SAME locktime and SAME heirs, so it stays valid
(it is NOT in the "clear VALID" list on purpose).
Other side effects:
* CONFIRMED, MEMPOOL -> clear INVALIDATED (the tx is on-chain or in
the mempool, so it is no longer considered invalidated).
* PUSHED -> clear PUSH_FAIL and CHECK_FAIL.
* CHECKED -> set PUSHED and clear PUSH_FAIL.
Args:
status: The status key to set (must exist in STATUS).
value: True to set the flag, False to clear it. Defaults to True.
Returns:
The applied boolean value, or None if the flag was already set to
that value (no change).
"""
if self.STATUS[status][1] == bool(value):
return None
self.status += "." + (("NOT " if not value else "") + _(self.STATUS[status][0]))
self.STATUS[status][1] = bool(value)
if value:
# NOTE: ANTICIPATED and UPDATED are intentionally NOT in this list,
# so they keep the VALID flag (see docstring above).
if status in ["INVALIDATED", "REPLACED", "CONFIRMED", "MEMPOOL"]:
self.STATUS["VALID"][1] = False
if status in ["CONFIRMED", "MEMPOOL"]:
self.STATUS["INVALIDATED"][1] = False
if status in ["PUSHED"]:
self.STATUS["PUSH_FAIL"][1] = False
self.STATUS["CHECK_FAIL"][1] = False
if status in ["CHECKED"]:
self.STATUS["PUSHED"][1] = True
self.STATUS["PUSH_FAIL"][1] = False
if status in ["COMPLETE"]:
self.STATUS["PARTIALLY_SIGNED"][1] = False
return value
def get_status(self, status):
return self.STATUS[status][1]
def __init__(self, w, _id=None, wallet=None):
if isinstance(
w,
WillItem,
):
self.__dict__ = w.__dict__.copy()
else:
self.tx = Will.get_tx_from_any(w["tx"])
self.heirs = w.get("heirs", None)
self.we = w.get("willexecutor", None)
self.status = w.get("status", None)
self.description = w.get("description", None)
self.time = w.get("time", None)
self.change = w.get("change", None)
self.tx_fees = w.get("baltx_fees", 0)
self.sigs_required = int(w.get("sigs_required", 0))
self.sigs_have = int(w.get("sigs_have", 0))
self.father = w.get("Father", None)
self.children = w.get("Children", None)
self.STATUS = copy.deepcopy(WillItem.STATUS_DEFAULT)
for s in self.STATUS:
self.STATUS[s][1] = w.get(s, WillItem.STATUS_DEFAULT[s][1])
# Backward-compatibility migration (A2): the "PENDING" status was
# renamed to "MEMPOOL". Wills saved by older versions of the plugin
# store the flag under the legacy "PENDING" key, so if that key is
# present and set, carry it over to "MEMPOOL". This way no state is
# lost when loading an older will. The new key always wins if both
# happen to be present.
if "MEMPOOL" not in w and w.get("PENDING"):
self.STATUS["MEMPOOL"][1] = True
if not _id:
self._id = self.tx.txid()
else:
self._id = _id
if not self._id:
self.status += "ERROR!!!"
self.valid = False
if wallet:
self.tx.add_info_from_wallet(wallet)
def to_dict(self):
out = {
"_id": self._id,
"tx": self.tx,
"heirs": self.heirs,
"willexecutor": self.we,
"status": self.status,
"description": self.description,
"time": self.time,
"change": self.change,
"baltx_fees": self.tx_fees,
"sigs_required": self.sigs_required,
"sigs_have": self.sigs_have,
}
for key in self.STATUS:
try:
out[key] = self.STATUS[key][1]
except Exception as e:
_logger.error(f"{key},{self.STATUS[key]} {e}")
return out
def __repr__(self):
return str(self)
def __str__(self):
return str(self.to_dict())
def set_anticipate(self, ow: "WillItem"):
nl = min(ow.tx.locktime, Will.check_anticipate(ow, self))
if int(nl) < self.tx.locktime:
self.tx.locktime = int(nl)
return True
else:
return False
def search_anticipate(self, all_inputs):
anticipated = False
for ow in self.search(all_inputs):
if self.set_anticipate(ow):
anticipated = True
return anticipated
def search(self, all_inputs):
for inp in self.tx.inputs():
prevout_str = inp.prevout.to_str()
oinps = all_inputs.get(prevout_str, [])
for oinp in oinps:
ow = oinp[1]
if ow._id != self._id:
yield ow
def normalize_locktime(self, all_inputs):
outputs = self.tx.outputs()
for idx in range(0, len(outputs)):
inps = all_inputs.get(f"{self._id}:{idx}", [])
_logger.debug("****check locktime***")
for inp in inps:
if inp[0] != self._id:
iw = inp[1]
self.set_anticipate(iw)
def set_check_willexecutor(self,resp):
try:
if resp :
if "tx" in resp and resp["tx"] == str(self.tx):
self.set_status("PUSHED")
self.set_status("CHECKED")
else:
self.set_status("CHECK_FAIL")
self.set_status("PUSHED", False)
return True
else:
self.set_status("CHECK_FAIL")
self.set_status("PUSHED", False)
return False
except Exception as e:
_logger.error(f"exception checking transaction: {e}")
self.set_status("CHECK_FAIL")
# NOTE: the former ``get_color()`` method (which returned hard-coded hex
# colours for the GUI) has been moved out of the core logic to
# ``bal.gui.qt.theme.status_color``. The status flags above remain the
# single source of truth; the GUI maps them to colours.
class WillException(Exception):
def __init__(self,msg="WillException"):
self.msg=msg
Exception.__init__(self)
def __str__(self):
return self.msg
class WillExpiredException(WillException):
pass
class NotCompleteWillException(WillException):
pass
class HeirChangeException(NotCompleteWillException):
pass
class TxFeesChangedException(NotCompleteWillException):
pass
class HeirNotFoundException(NotCompleteWillException):
pass
class WillPostponedException(NotCompleteWillException):
"""An already signed/sent will is being postponed.
When a will that has already been signed (``COMPLETE``) and/or pushed to
will-executors (``PUSHED``) gets its locktime moved to a LATER date, the
previously committed coins must be invalidated on-chain BEFORE rebuilding
the new inheritance. Otherwise a will-executor could broadcast the old
(earlier-locktime) transaction and execute the inheritance too early to
collect the fees. Invalidating spends the same UTXOs now, permanently
voiding the old pre-signed transaction.
"""
pass
class WillexecutorChangeException(NotCompleteWillException):
pass
class NoWillExecutorNotPresent(NotCompleteWillException):
pass
class WillExecutorNotPresent(NotCompleteWillException):
pass
class NoHeirsException(WillException):
pass
class AmountException(WillException):
pass
class PercAmountException(AmountException):
pass
class FixedAmountException(AmountException):
pass