"""
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 (
bfh,
)
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_inputs)
will[wid] = WillItem(ow.to_dict())
for i in range(0, len(outputs)):
Will.change_input(
will, wid, i, outputs[i], others_inputs, 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
else:
if int(ow.tx_fees) != int(nw.tx_fees):
return anticipate
else:
ow.tx.locktime
else:
ow.tx.locktime
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
for inp in wtx.inputs():
if inp.prevout.txid.hex() in will:
found = True
break
if not found:
out[inp.prevout.to_str()] = inp
return out
@staticmethod
def invalidate_will(will, wallet, fees_per_byte):
will_only_valid = Will.only_valid_list(will)
inputs = Will.get_all_inputs(will_only_valid)
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:
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)
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:
if wallet.db.get_transaction(wi._id):
wi.set_status("CONFIRMED", True)
else:
wi.set_status("INVALIDATED", 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 wallet:
height = Will.check_tx_height(w.tx, wallet)
if height < 0:
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):
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):
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 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.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)
@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 prevout_str, 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):
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],
"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
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.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,
}
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