SAFE21.io/bal-electrum-plugin
1
0
Fork 0
forked from bitcoinafterlife/bal-electrum-plugin
Code Pull Requests Activity
Files
50129d3b77bcd5b27d7bcb9e114599691ec3c2bd
bal-electrum-plugin/bal/core/will.py
donkey-ai dc166d04ff feat(bal): Group A (timestamps + statuses + anticipate docs) and Group B (auto-sign) v0.3.4 
Bump plugin version to 0.3.4 (manifest, __init__, plugin_base, VERSION).

GROUP A
- A1: remove block-height locktimes; the plugin now uses UNIX timestamps
  only. The NLOCKTIME_BLOCKHEIGHT_MAX guard is kept on purpose (it forces
  every locktime to be a timestamp). chk_locktime is now 2-arg; int_locktime
  and anticipate_locktime no longer accept blocks; RAW input only accepts d/y.
  Two now-dormant configs (LOCKTIME_BLOCKS, LOCKTIMEDELTA_BLOCKS) are kept with
  comments to avoid touching persisted keys.
- A2: rename PENDING -> MEMPOOL everywhere (label 'Mempool', yellow #ffce30);
  add new UPDATED status; ANTICIPATED & UPDATED keep VALID; documented
  set_status rules; backward-compat migration (old PENDING -> MEMPOOL).
- A3: clarify that anticipating to a future date only rebuilds (never
  invalidates), while only a past locktime invalidates (WillExpired). Code was
  already correct; only the comment and docs were fixed.

Colour follow-up: UPDATED lightened from #800080 to #b266b2 (more readable),
updated in theme.py, docs and the theme test.

GROUP B
- B1: verified the 'Create your will' button already opens the guided wizard
  (no code change needed).
- B2: new persisted AUTO_SIGN setting (default ON) with an 'Auto-sign on Check'
  checkbox in the settings dialog. When enabled, Check signs and broadcasts
  automatically; the wallet password is requested only for encrypted wallets.

B2 follow-up (fixes reported after testing):
- Remove the duplicate sign/broadcast cycle in lists.check(); build_will_task()
  already signs and broadcasts.
- Suppress the manual 'press Sign/Broadcast' hint and its popup when AUTO_SIGN
  is ON (kept when OFF).
- Make broadcast one-shot: removed the retry flag and the Exception('retry');
  failed will-executors stay PUSH_FAIL and are skipped (no endless retry).
  PUSHED transactions are already excluded from re-collection.

Docs: inheritance-options.md/.html and inheritance-flow.svg updated to v0.3.4.
Tests: 206 passing (new test_anticipate_manual_locktime, test_anticipate_past_locktime,
test_group_b_auto_sign; updated core/util, core/will_extra, gui/theme, gui/widgets).
CHANGELOG.md added with one numbered entry per task.
2026-06-28 23:00:23 -04:00

1105 lines
42 KiB
Python
Raw Blame History

"""
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 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 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
new_will[oid] = old_will[oid]
new_will[oid].heirs = new_heirs
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 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``.
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):
raise WillExpiredException(
f"Will Expired {wid[0][0]}: {locktime}<{timestamp_to_check}"
)
else:
from datetime import datetime
_logger.debug(f"Will Not Expired {wid[0][0]}: {datetime.fromtimestamp(locktime).isoformat()} > {datetime.fromtimestamp(timestamp_to_check).isoformat()}")
# 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
View Git Blame Copy Permalink