791 lines
29 KiB
Python
791 lines
29 KiB
Python
"""
|
|
bal.core.util
|
|
=============
|
|
|
|
Small, stateless helper functions shared across the whole plugin.
|
|
|
|
This module is intentionally GUI-free: it only deals with locktimes, amount
|
|
encoding/decoding, and comparing transactions / inputs / outputs / heirs.
|
|
|
|
Historical note
|
|
---------------
|
|
The original ``util.py`` also contained a set of ``print_*`` debugging helpers
|
|
(``print_var``, ``print_utxo``, ``print_prevout``) that dumped objects to
|
|
``stdout``. Those were development-only scaffolding, never called by the
|
|
plugin logic, so they have been removed during the rewrite. No behavioural
|
|
function has been changed: every method below is logically identical to the
|
|
original implementation.
|
|
"""
|
|
|
|
import bisect
|
|
from datetime import datetime, timedelta
|
|
|
|
from electrum.address_synchronizer import TX_HEIGHT_FUTURE, TX_HEIGHT_LOCAL
|
|
from electrum.transaction import PartialTxOutput
|
|
|
|
# Bitcoin consensus rule: an nLockTime value strictly below this threshold is
|
|
# interpreted as a *block height*, otherwise it is interpreted as a *UNIX
|
|
# timestamp*.
|
|
#
|
|
# The plugin now uses ONLY timestamp-based locktimes (block-height locktimes
|
|
# were removed so that every locktime can be compared and ordered consistently).
|
|
# This constant is kept as a guard: it is the boundary that lets us reject any
|
|
# value that would fall in the block-height range and force every locktime to be
|
|
# a timestamp.
|
|
LOCKTIME_THRESHOLD = 500000000
|
|
|
|
|
|
class Util:
|
|
"""Namespace of static helpers (kept as a class to preserve the original
|
|
``Util.method(...)`` call sites used throughout the plugin)."""
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# Locktime helpers
|
|
# ------------------------------------------------------------------ #
|
|
@staticmethod
|
|
def locktime_to_str(locktime):
|
|
"""Render a locktime for display.
|
|
|
|
If the value looks like a timestamp (``> LOCKTIME_THRESHOLD``) it is
|
|
formatted as an ISO date string; otherwise it is returned as-is.
|
|
"""
|
|
try:
|
|
locktime = int(locktime)
|
|
if locktime > LOCKTIME_THRESHOLD:
|
|
dt = datetime.fromtimestamp(locktime).isoformat()
|
|
return dt
|
|
|
|
except Exception:
|
|
pass
|
|
return str(locktime)
|
|
|
|
@staticmethod
|
|
def str_to_locktime(locktime):
|
|
"""Parse a user-entered locktime string into its stored form.
|
|
|
|
Relative values keep their suffix (``"30d"``, ``"1y"``); absolute ISO
|
|
dates are converted to an integer UNIX timestamp.
|
|
|
|
Note: only timestamp-based locktimes are supported. The legacy
|
|
block-height suffix ``"b"`` has been removed on purpose, so that every
|
|
locktime in the plugin is a UNIX timestamp and can always be compared
|
|
and ordered consistently.
|
|
"""
|
|
try:
|
|
if locktime[-1] in ("y", "d"):
|
|
return locktime
|
|
else:
|
|
return int(locktime)
|
|
except Exception:
|
|
pass
|
|
dt_object = datetime.fromisoformat(locktime)
|
|
timestamp = dt_object.timestamp()
|
|
return int(timestamp)
|
|
|
|
@staticmethod
|
|
def parse_locktime_string(locktime, w=None):
|
|
"""Resolve a (possibly relative) locktime string into a concrete int.
|
|
|
|
Supported forms:
|
|
* plain int / timestamp -> returned unchanged
|
|
* ``"<n>y"`` -> n years from now (as a timestamp)
|
|
* ``"<n>d"`` -> n days from now (as a timestamp)
|
|
|
|
Note: the legacy block-height form ``"<n>b"`` has been removed on
|
|
purpose. Every locktime is now a UNIX timestamp, so locktimes can
|
|
always be compared and ordered consistently. The optional ``w``
|
|
(wallet) argument is kept only for backward call-site compatibility and
|
|
is no longer used.
|
|
"""
|
|
try:
|
|
return int(locktime)
|
|
|
|
except Exception:
|
|
pass
|
|
try:
|
|
now = datetime.now()
|
|
if locktime[-1] == "y":
|
|
locktime = str(int(locktime[:-1]) * 365) + "d"
|
|
if locktime[-1] == "d":
|
|
return int(
|
|
(now + timedelta(days=int(locktime[:-1])))
|
|
.replace(hour=0, minute=0, second=0, microsecond=0)
|
|
.timestamp()
|
|
)
|
|
return int(locktime)
|
|
except Exception:
|
|
pass
|
|
return 0
|
|
|
|
@staticmethod
|
|
def int_locktime(seconds=0, minutes=0, hours=0, days=0):
|
|
"""Convert a human duration into seconds.
|
|
|
|
Note: the ``blocks`` argument was removed together with block-height
|
|
support; every duration is now expressed in plain time units.
|
|
"""
|
|
return int(
|
|
seconds
|
|
+ minutes * 60
|
|
+ hours * 60 * 60
|
|
+ days * 60 * 60 * 24
|
|
)
|
|
|
|
@staticmethod
|
|
def _relative_days(value):
|
|
"""Duration in days of a relative ``"Nd"``/``"Ny"`` recipe.
|
|
|
|
Returns ``None`` when the value is not a relative recipe (an absolute
|
|
timestamp, a plain number, or garbage).
|
|
"""
|
|
s = str(value)
|
|
if s and s[-1] in "yYdD":
|
|
try:
|
|
n = int(s[:-1])
|
|
except ValueError:
|
|
return None
|
|
return n * 365 if s[-1] in "yY" else n
|
|
return None
|
|
|
|
@staticmethod
|
|
def resolve_locktime_against_tx(current, built, tx_locktime):
|
|
"""Resolve a locktime recipe against the moment the signed tx was built.
|
|
|
|
A RELATIVE recipe stored in the wallet (``"1y"``/``"30d"``) is a moving
|
|
target: parsing it against *now* on every check drifts it one day per
|
|
day away from the fixed locktime frozen inside the signed Bitcoin
|
|
transaction, so an UNCHANGED will is mistaken for a POSTPONE and the
|
|
plugin asks to invalidate it every day (reported bug). This resolves
|
|
the current recipe against the build moment instead, recovered from the
|
|
signed transaction's locktime and the recipe that was actually frozen
|
|
at build time (``built``, the value stored in the will item):
|
|
|
|
build_moment = tx_locktime - duration(built)
|
|
expected = build_moment + duration(current)
|
|
|
|
An unchanged recipe therefore resolves to exactly ``tx_locktime``
|
|
(coherent), a lengthened one resolves later (postpone) and a shortened
|
|
one earlier (anticipate).
|
|
|
|
Args:
|
|
current: the current locktime recipe (relative or absolute).
|
|
built: the recipe frozen at build time (stored in the will item).
|
|
tx_locktime: the absolute locktime frozen inside the signed tx.
|
|
|
|
Returns:
|
|
int: the resolved absolute locktime (UNIX timestamp).
|
|
"""
|
|
current_days = Util._relative_days(current)
|
|
built_days = Util._relative_days(built)
|
|
if current_days is None:
|
|
# Absolute current date: compare directly against the frozen tx.
|
|
try:
|
|
return int(current)
|
|
except Exception:
|
|
return Util.parse_locktime_string(current)
|
|
if built_days is None or not tx_locktime:
|
|
# The stored recipe was absolute (a fixed date) or the tx has no
|
|
# usable locktime: there is no relative anchor to recover the build
|
|
# moment, so fall back to the legacy forward-from-now resolution.
|
|
return Util.parse_locktime_string(current)
|
|
try:
|
|
base = datetime.fromtimestamp(int(tx_locktime)).replace(
|
|
hour=0, minute=0, second=0, microsecond=0
|
|
)
|
|
build_moment = base - timedelta(days=built_days)
|
|
return int((build_moment + timedelta(days=current_days)).timestamp())
|
|
except Exception:
|
|
return Util.parse_locktime_string(current)
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# Amount helpers
|
|
# ------------------------------------------------------------------ #
|
|
@staticmethod
|
|
def encode_amount(amount, decimal_point):
|
|
"""Convert a displayed BTC amount into integer satoshis.
|
|
|
|
Percentage amounts (e.g. ``"50%"``) are passed through unchanged, since
|
|
they are resolved later against the wallet balance.
|
|
"""
|
|
if Util.is_perc(amount):
|
|
return amount
|
|
else:
|
|
try:
|
|
return int(float(amount) * pow(10, decimal_point))
|
|
except Exception:
|
|
return 0
|
|
|
|
@staticmethod
|
|
def decode_amount(amount, decimal_point):
|
|
"""Inverse of :meth:`encode_amount`: satoshis -> displayed string."""
|
|
if Util.is_perc(amount):
|
|
return amount
|
|
else:
|
|
basestr = "{{:0.{}f}}".format(decimal_point)
|
|
try:
|
|
return basestr.format(float(amount) / pow(10, decimal_point))
|
|
except Exception:
|
|
return str(amount)
|
|
|
|
@staticmethod
|
|
def is_perc(value):
|
|
"""True if ``value`` is a percentage string such as ``"25%"``."""
|
|
try:
|
|
return value[-1] == "%"
|
|
except Exception:
|
|
return False
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# Heir / will-executor comparison helpers
|
|
# ------------------------------------------------------------------ #
|
|
@staticmethod
|
|
def cmp_array(heira, heirb):
|
|
"""Element-wise equality of two sequences (length-safe)."""
|
|
try:
|
|
if len(heira) != len(heirb):
|
|
return False
|
|
for h in range(0, len(heira)):
|
|
if heira[h] != heirb[h]:
|
|
return False
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
@staticmethod
|
|
def cmp_heir(heira, heirb):
|
|
"""Two heirs are "the same" when address (0) and amount (1) match."""
|
|
if heira[0] == heirb[0] and heira[1] == heirb[1]:
|
|
return True
|
|
return False
|
|
|
|
@staticmethod
|
|
def cmp_willexecutor(willexecutora, willexecutorb):
|
|
"""Compare two will-executor dicts by url / address / base_fee."""
|
|
if willexecutora == willexecutorb:
|
|
return True
|
|
try:
|
|
if (
|
|
willexecutora["url"] == willexecutorb["url"]
|
|
and willexecutora["address"] == willexecutorb["address"]
|
|
and willexecutora["base_fee"] == willexecutorb["base_fee"]
|
|
):
|
|
return True
|
|
except Exception:
|
|
return False
|
|
return False
|
|
|
|
@staticmethod
|
|
def search_heir_by_values(heirs, heir, values):
|
|
"""Return the key of the first heir in ``heirs`` matching ``heir`` on
|
|
every column listed in ``values`` (or ``False`` if none)."""
|
|
for h, v in heirs.items():
|
|
found = False
|
|
for val in values:
|
|
if val in v and v[val] != heir[val]:
|
|
found = True
|
|
|
|
if not found:
|
|
return h
|
|
return False
|
|
|
|
@staticmethod
|
|
def cmp_heir_by_values(heira, heirb, values):
|
|
"""True when two heirs agree on every column index in ``values``."""
|
|
for v in values:
|
|
if heira[v] != heirb[v]:
|
|
return False
|
|
return True
|
|
|
|
@staticmethod
|
|
def cmp_heirs_by_values(
|
|
heirsa, heirsb, values, exclude_willexecutors=False, reverse=True
|
|
):
|
|
"""Set-equality of two heir collections, comparing only ``values``.
|
|
|
|
When ``exclude_willexecutors`` is set, synthetic will-executor heirs
|
|
(those whose key contains the ``w!ll3x3c"`` marker) are skipped. The
|
|
``reverse`` flag makes the comparison symmetric by running it both ways.
|
|
"""
|
|
for heira in heirsa:
|
|
if (
|
|
exclude_willexecutors and 'w!ll3x3c"' not in heira
|
|
) or not exclude_willexecutors:
|
|
found = False
|
|
for heirb in heirsb:
|
|
if Util.cmp_heir_by_values(heirsa[heira], heirsb[heirb], values):
|
|
found = True
|
|
if not found:
|
|
return False
|
|
if reverse:
|
|
return Util.cmp_heirs_by_values(
|
|
heirsb,
|
|
heirsa,
|
|
values,
|
|
exclude_willexecutors=exclude_willexecutors,
|
|
reverse=False,
|
|
)
|
|
else:
|
|
return True
|
|
|
|
@staticmethod
|
|
def cmp_heirs(
|
|
heirsa,
|
|
heirsb,
|
|
cmp_function=lambda x, y: x[0] == y[0] and x[3] == y[3],
|
|
reverse=True,
|
|
):
|
|
"""Compare two heir collections using a custom ``cmp_function``.
|
|
|
|
Will-executor entries are ignored. As with
|
|
:meth:`cmp_heirs_by_values`, ``reverse`` makes the relation symmetric.
|
|
"""
|
|
try:
|
|
for heir in heirsa:
|
|
if 'w!ll3x3c"' not in heir:
|
|
if heir not in heirsb or not cmp_function(
|
|
heirsa[heir], heirsb[heir]
|
|
):
|
|
if not Util.search_heir_by_values(heirsb, heirsa[heir], [0, 3]):
|
|
return False
|
|
if reverse:
|
|
return Util.cmp_heirs(heirsb, heirsa, cmp_function, False)
|
|
else:
|
|
return True
|
|
except Exception as e:
|
|
raise e
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# Transaction input/output comparison helpers
|
|
# ------------------------------------------------------------------ #
|
|
@staticmethod
|
|
def cmp_inputs(inputsa, inputsb):
|
|
"""True when both input lists reference the same set of UTXOs."""
|
|
if len(inputsa) != len(inputsb):
|
|
return False
|
|
for inputa in inputsa:
|
|
if not Util.in_utxo(inputa, inputsb):
|
|
return False
|
|
return True
|
|
|
|
@staticmethod
|
|
def cmp_outputs(outputsa, outputsb, willexecutor_output=None):
|
|
"""True when both output lists contain the same (address, value) pairs.
|
|
|
|
The optional ``willexecutor_output`` is treated as a wildcard match so
|
|
that the will-executor's fee output does not break the comparison.
|
|
"""
|
|
if len(outputsa) != len(outputsb):
|
|
return False
|
|
for outputa in outputsa:
|
|
if not Util.cmp_output(outputa, willexecutor_output):
|
|
if not Util.in_output(outputa, outputsb):
|
|
return False
|
|
return True
|
|
|
|
@staticmethod
|
|
def cmp_txs(txa, txb):
|
|
"""Two transactions are equivalent when their inputs and outputs match."""
|
|
if not Util.cmp_inputs(txa.inputs(), txb.inputs()):
|
|
return False
|
|
if not Util.cmp_outputs(txa.outputs(), txb.outputs()):
|
|
return False
|
|
return True
|
|
|
|
@staticmethod
|
|
def get_value_amount(txa, txb):
|
|
"""Sum of the values of outputs that appear (same addr+value) in both
|
|
transactions. Returns ``False`` as soon as an output of ``txa`` shares
|
|
neither amount nor address with any output of ``txb``."""
|
|
outputsa = txa.outputs()
|
|
value_amount = 0
|
|
|
|
for outa in outputsa:
|
|
same_amount, same_address = Util.din_output(outa, txb.outputs())
|
|
if not (same_amount or same_address):
|
|
return False
|
|
if same_amount and same_address:
|
|
value_amount += outa.value
|
|
if same_amount:
|
|
pass
|
|
if same_address:
|
|
pass
|
|
|
|
return value_amount
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# Locktime arithmetic
|
|
# ------------------------------------------------------------------ #
|
|
@staticmethod
|
|
def chk_locktime(timestamp_to_check, locktime):
|
|
"""Return True if ``locktime`` (a UNIX timestamp) is still in the future.
|
|
|
|
Only timestamp-based locktimes are supported now; the previous
|
|
block-height branch was removed together with block-height support.
|
|
"""
|
|
locktime = int(locktime)
|
|
return locktime > int(timestamp_to_check)
|
|
|
|
@staticmethod
|
|
def anticipate_locktime(locktime, hours=0, days=0):
|
|
"""Move a timestamp locktime earlier by the given amount.
|
|
|
|
Every locktime is a UNIX timestamp now, so this simply subtracts the
|
|
requested time span. The result is never allowed to drop below 1.
|
|
|
|
Note: the legacy ``blocks`` argument and the block-height branch were
|
|
removed; only timestamp arithmetic remains.
|
|
"""
|
|
locktime = int(locktime)
|
|
seconds = hours * 3600 + days * 86400
|
|
# On Windows datetime.fromtimestamp raises OverflowError past 2038
|
|
# (e.g. NLOCKTIME_MAX); clamp to INT32_MAX (Electrum issue #6170).
|
|
try:
|
|
dt = datetime.fromtimestamp(locktime)
|
|
except (OverflowError, OSError, ValueError):
|
|
dt = datetime.fromtimestamp(min(locktime, 2 ** 31 - 1))
|
|
dt -= timedelta(seconds=seconds)
|
|
out = dt.timestamp()
|
|
|
|
if out < 1:
|
|
out = 1
|
|
return out
|
|
|
|
@staticmethod
|
|
def cmp_locktime(locktimea, locktimeb):
|
|
"""Compare two relative locktime strings sharing the same unit."""
|
|
if locktimea == locktimeb:
|
|
return 0
|
|
strlocktimea = str(locktimea)
|
|
strlocktimeb = str(locktimeb)
|
|
if locktimea[-1] in "ydb":
|
|
if locktimeb[-1] == locktimea[-1]:
|
|
return int(strlocktimea[-1]) - int(strlocktimeb[-1])
|
|
else:
|
|
return int(locktimea) - (locktimeb)
|
|
|
|
@staticmethod
|
|
def get_lowest_valid_tx(available_utxos, will):
|
|
"""Placeholder kept from the original code (sorts the will by locktime)."""
|
|
will = sorted(will.items(), key=lambda x: x[1]["tx"].locktime)
|
|
for _txid, _willitem in will.items():
|
|
pass
|
|
|
|
@staticmethod
|
|
def get_locktimes(will):
|
|
"""Return the distinct locktimes used by the transactions in ``will``."""
|
|
locktimes = {}
|
|
for _, willitem in will.items():
|
|
locktimes[willitem["tx"].locktime] = True
|
|
return locktimes.keys()
|
|
|
|
@staticmethod
|
|
def get_lowest_locktimes(locktimes):
|
|
"""Split a list of locktimes into (sorted_timestamps, sorted_blocks)."""
|
|
sorted_timestamp = []
|
|
sorted_block = []
|
|
for locktime in locktimes:
|
|
locktime = Util.parse_locktime_string(locktime)
|
|
if locktime < LOCKTIME_THRESHOLD:
|
|
bisect.insort(sorted_block, locktime)
|
|
else:
|
|
bisect.insort(sorted_timestamp, locktime)
|
|
|
|
return sorted(sorted_timestamp), sorted(sorted_block)
|
|
|
|
@staticmethod
|
|
def get_lowest_locktimes_from_will(will):
|
|
"""Convenience wrapper: lowest locktimes directly from a will dict."""
|
|
return Util.get_lowest_locktimes(Util.get_locktimes(will))
|
|
|
|
@staticmethod
|
|
def search_willtx_per_io(will, tx):
|
|
"""Find a will entry whose tx has the same inputs/outputs as ``tx``."""
|
|
for wid, w in will.items():
|
|
if Util.cmp_txs(w["tx"], tx["tx"]):
|
|
return wid, w
|
|
return None, None
|
|
|
|
@staticmethod
|
|
def invalidate_will(will):
|
|
raise Exception("not implemented")
|
|
|
|
@staticmethod
|
|
def get_will_spent_utxos(will):
|
|
"""Collect every input spent by any transaction in ``will``."""
|
|
utxos = []
|
|
for _, willitem in will.items():
|
|
utxos += willitem["tx"].inputs()
|
|
|
|
return utxos
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# UTXO helpers
|
|
# ------------------------------------------------------------------ #
|
|
@staticmethod
|
|
def utxo_to_str(utxo):
|
|
"""Best-effort conversion of a UTXO / input object to its ``txid:n`` str."""
|
|
try:
|
|
return utxo.to_str()
|
|
except Exception:
|
|
pass
|
|
try:
|
|
return utxo.prevout.to_str()
|
|
except Exception:
|
|
pass
|
|
return str(utxo)
|
|
|
|
@staticmethod
|
|
def cmp_utxo(utxoa, utxob):
|
|
"""True when two UTXOs refer to the same outpoint."""
|
|
utxoa = Util.utxo_to_str(utxoa)
|
|
utxob = Util.utxo_to_str(utxob)
|
|
if utxoa == utxob:
|
|
return True
|
|
else:
|
|
return False
|
|
|
|
@staticmethod
|
|
def in_utxo(utxo, utxos):
|
|
"""Membership test for a UTXO inside an iterable of UTXOs."""
|
|
for s_u in utxos:
|
|
if Util.cmp_utxo(s_u, utxo):
|
|
return True
|
|
return False
|
|
|
|
@staticmethod
|
|
def txid_in_utxo(txid, utxos):
|
|
"""True if any UTXO in ``utxos`` is spent from transaction ``txid``."""
|
|
for s_u in utxos:
|
|
if s_u.prevout.txid == txid:
|
|
return True
|
|
return False
|
|
|
|
@staticmethod
|
|
def get_available_utxos(wallet, history_label, will_locktime=None):
|
|
"""Return the wallet's UTXOs as seen by the plugin's flows.
|
|
|
|
``wallet.get_utxos()`` drops any output that a wallet-LOCAL transaction
|
|
marks as spent. The plugin itself creates such local spenders when it
|
|
saves an incomplete will transaction into the local history; a *later*
|
|
will transaction stored there (a replacement/future will with a locktime
|
|
strictly after ``will_locktime``) must not hide the coins from the will
|
|
being checked or rebuilt. This view therefore restores those coins.
|
|
|
|
A local spender is ignored (the coin is kept available) only when ALL of
|
|
these hold:
|
|
|
|
* it is a wallet-local or future transaction (not broadcast),
|
|
* its wallet label matches the BAL history label template (after the
|
|
"{willexecutor}" substitution),
|
|
* the stored spender's locktime is strictly LATER than ``will_locktime``.
|
|
|
|
Real (broadcast/confirmed) spenders are never ignored. With a falsy
|
|
``will_locktime`` this returns ``wallet.get_utxos()`` unchanged.
|
|
|
|
Args:
|
|
wallet: The Electrum wallet object.
|
|
history_label: The BAL history label template (may contain
|
|
"{willexecutor}").
|
|
will_locktime: Reference locktime of the will being operated on.
|
|
"""
|
|
if not wallet or not will_locktime:
|
|
return list(wallet.get_utxos()) if wallet else []
|
|
adb = getattr(wallet, "adb", None)
|
|
if adb is None or not hasattr(adb, "get_addr_outputs"):
|
|
return list(wallet.get_utxos())
|
|
addresses = (
|
|
wallet.get_addresses() if hasattr(wallet, "get_addresses") else []
|
|
)
|
|
utxos = []
|
|
for addr in addresses:
|
|
try:
|
|
outputs = adb.get_addr_outputs(addr)
|
|
except Exception:
|
|
continue
|
|
for utxo in outputs.values():
|
|
if utxo.spent_height is None:
|
|
utxos.append(utxo)
|
|
continue
|
|
spender = getattr(utxo, "spent_txid", None)
|
|
if spender and Util._is_ignorable_local_spender(
|
|
wallet, spender, history_label, will_locktime
|
|
):
|
|
utxos.append(utxo)
|
|
return utxos
|
|
|
|
@staticmethod
|
|
def _is_ignorable_local_spender(wallet, spender, history_label, will_locktime):
|
|
"""True when the local ``spender`` tx is a later BAL history will tx.
|
|
|
|
See ``get_available_utxos`` for the exact conditions. Defensive: any
|
|
lookup failure makes this return False, so a spender is never ignored
|
|
on uncertain data.
|
|
"""
|
|
adb = wallet.adb
|
|
try:
|
|
height = int(adb.get_tx_height(spender).height())
|
|
except Exception:
|
|
return False
|
|
if height not in (TX_HEIGHT_LOCAL, TX_HEIGHT_FUTURE):
|
|
return False
|
|
try:
|
|
label = wallet.get_label_for_txid(spender)
|
|
except Exception:
|
|
label = None
|
|
if not label or not Util._label_matches_history(label, history_label):
|
|
return False
|
|
try:
|
|
stored = adb.db.get_transaction(spender)
|
|
except Exception:
|
|
return False
|
|
if stored is None:
|
|
return False
|
|
try:
|
|
return int(stored.locktime) > int(will_locktime)
|
|
except Exception:
|
|
return False
|
|
|
|
@staticmethod
|
|
def _label_matches_history(label, history_label):
|
|
"""True when ``label`` is the ``history_label`` template with the
|
|
"{willexecutor}" token substituted by some (possibly empty) executor URL.
|
|
"""
|
|
token = "{willexecutor}"
|
|
if token in history_label:
|
|
prefix, suffix = history_label.split(token, 1)
|
|
return (
|
|
label.startswith(prefix)
|
|
and label.endswith(suffix)
|
|
and len(label) >= len(prefix) + len(suffix)
|
|
)
|
|
return label == history_label
|
|
|
|
@staticmethod
|
|
def cmp_output(outputa, outputb):
|
|
"""Two outputs are equal when both address and value match."""
|
|
return outputa.address == outputb.address and outputa.value == outputb.value
|
|
|
|
@staticmethod
|
|
def in_output(output, outputs):
|
|
"""Membership test for an output inside an iterable of outputs."""
|
|
for s_o in outputs:
|
|
if Util.cmp_output(s_o, output):
|
|
return True
|
|
return False
|
|
|
|
# check all output with the same amount if none have the same address it can be a change
|
|
# return true true same address same amount
|
|
# return true false same amount different address
|
|
# return false false different amount, different address not found
|
|
@staticmethod
|
|
def din_output(out, outputs):
|
|
"""Detailed output lookup used to tell a change output apart.
|
|
|
|
Returns a ``(same_amount, same_address)`` tuple:
|
|
* ``(True, True)`` -> an output with same amount *and* address
|
|
* ``(True, False)`` -> same amount but different address (maybe change)
|
|
* ``(False, False)``-> no output with this amount
|
|
"""
|
|
same_amount = []
|
|
for s_o in outputs:
|
|
if int(out.value) == int(s_o.value):
|
|
same_amount.append(s_o)
|
|
if out.address == s_o.address:
|
|
return True, True
|
|
else:
|
|
pass
|
|
|
|
if len(same_amount) > 0:
|
|
return True, False
|
|
else:
|
|
return False, False
|
|
|
|
@staticmethod
|
|
def get_change_output(wallet, in_amount, out_amount, fee):
|
|
"""Build a change ``PartialTxOutput`` if the leftover exceeds dust."""
|
|
change_amount = int(in_amount - out_amount - fee)
|
|
if change_amount > wallet.dust_threshold():
|
|
change_addresses = wallet.get_change_addresses_for_new_transaction()
|
|
out = PartialTxOutput.from_address_and_value(
|
|
change_addresses[0], change_amount
|
|
)
|
|
out.is_change = True
|
|
return out
|
|
|
|
@staticmethod
|
|
def get_current_height(network):
|
|
"""Return a conservative current block height for locktime purposes.
|
|
|
|
Mirrors Electrum's own anti-fee-sniping logic: if there is no network,
|
|
the chain tip is stale, or the main server lags too far behind the
|
|
SPV-checked height, it gives up and returns 0.
|
|
"""
|
|
# if no network or not up to date, just set locktime to zero
|
|
if not network:
|
|
return 0
|
|
chain = network.blockchain()
|
|
if chain.is_tip_stale():
|
|
return 0
|
|
# figure out current block height
|
|
chain_height = chain.height() # learnt from all connected servers, SPV-checked
|
|
server_height = (
|
|
network.get_server_height()
|
|
) # height claimed by main server, unverified
|
|
# note: main server might be lagging (either is slow, is malicious, or there is an SPV-invisible-hard-fork)
|
|
# - if it's lagging too much, it is the network's job to switch away
|
|
if server_height < chain_height - 10:
|
|
# the diff is suspiciously large... give up and use something non-fingerprintable
|
|
return 0
|
|
# discourage "fee sniping"
|
|
height = min(chain_height, server_height)
|
|
return height
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# Misc helpers
|
|
# ------------------------------------------------------------------ #
|
|
@staticmethod
|
|
def copy(dicto, dictfrom):
|
|
"""Shallow copy of ``dictfrom`` entries into ``dicto`` (in place)."""
|
|
for k, v in dictfrom.items():
|
|
dicto[k] = v
|
|
|
|
@staticmethod
|
|
def fix_will_settings_tx_fees(will_settings):
|
|
"""Migrate the legacy ``tx_fees`` key to ``baltx_fees`` in settings.
|
|
|
|
Returns True when a migration was performed (caller should persist).
|
|
"""
|
|
tx_fees = will_settings.get("tx_fees", False)
|
|
have_to_update = False
|
|
if tx_fees:
|
|
will_settings["baltx_fees"] = tx_fees
|
|
del will_settings["tx_fees"]
|
|
have_to_update = True
|
|
return have_to_update
|
|
|
|
@staticmethod
|
|
def fix_will_tx_fees(will):
|
|
"""Same legacy migration as above but applied to every will entry."""
|
|
have_to_update = False
|
|
for txid, willitem in will.items():
|
|
tx_fees = willitem.get("tx_fees", False)
|
|
if tx_fees:
|
|
will[txid]["baltx_fees"] = tx_fees
|
|
del will[txid]["tx_fees"]
|
|
have_to_update = True
|
|
return have_to_update
|
|
|
|
@staticmethod
|
|
def text_to_hex(text: str) -> str:
|
|
"""Convert text to a hexadecimal string (used for OP_RETURN payloads)."""
|
|
hex_string = text.encode('utf-8').hex()
|
|
return hex_string
|
|
|
|
@staticmethod
|
|
def hex_to_text(hex_string: str) -> str:
|
|
"""Convert a hexadecimal string back to text (for verification)."""
|
|
try:
|
|
return bytes.fromhex(hex_string).decode('utf-8')
|
|
except Exception:
|
|
return "Error: Invalid hex string"
|