Browse Source

wallet: refactor check_sighash/risk_of_burning_coins_as_fees

- risk_of_burning_coins_as_fees is turned into a private (helper) method, only called by check_sighash.
  UIs should only care about check_sighash.
- check_sighash returns instance of new class "TxSighashDanger" instead of tuple
- made warning levels more fine-grained (FEE_WARNING_SKIPCONFIRM vs FEE_WARNING_NEEDCONFIRM)
- this became more complicated than I had hoped for but I think it is worth it to ~merge
  check_sighash and risk_of_burning_coins_as_fees into one.

related https://github.com/spesmilo/electrum/pull/8699
master
SomberNight 2 years ago
parent
commit
2bc056ed33
No known key found for this signature in database
GPG Key ID: B33B5F232C6271E9
  1. 2
      electrum/gui/qml/components/TxDetails.qml
  2. 27
      electrum/gui/qml/qetxdetails.py
  3. 41
      electrum/gui/qt/transaction_dialog.py
  4. 7
      electrum/tests/test_wallet_vertical.py
  5. 180
      electrum/wallet.py

2
electrum/gui/qml/components/TxDetails.qml

@ -362,7 +362,7 @@ Pane {
onClicked: {
if (txdetails.shouldConfirm) {
var dialog = app.messageDialog.createObject(app, {
text: qsTr('Confirm signing non-standard transaction?'),
text: qsTr('Confirm signing transaction despite warnings?'),
yesno: true
})
dialog.accepted.connect(function() {

27
electrum/gui/qml/qetxdetails.py

@ -8,6 +8,7 @@ from electrum.util import format_time, TxMinedInfo
from electrum.transaction import tx_from_any, Transaction, PartialTxInput, Sighash, PartialTransaction
from electrum.network import Network
from electrum.address_synchronizer import TX_HEIGHT_UNCONF_PARENT, TX_HEIGHT_UNCONFIRMED, TX_HEIGHT_FUTURE
from electrum.wallet import TxSighashDanger
from .qewallet import QEWallet
from .qetypes import QEAmount
@ -32,7 +33,6 @@ class QETxDetails(QObject, QtEventListener):
self._txid = ''
self._rawtx = ''
self._label = ''
self._warning = ''
self._tx = None # type: Optional[Transaction]
@ -57,7 +57,7 @@ class QETxDetails(QObject, QtEventListener):
self._is_mined = False
self._is_final = False
self._lock_delay = 0
self._should_confirm = False
self._sighash_danger = TxSighashDanger()
self._mempool_depth = ''
self._in_mempool = False
@ -145,7 +145,7 @@ class QETxDetails(QObject, QtEventListener):
@pyqtProperty(str, notify=detailsChanged)
def warning(self):
return self._warning
return self._sighash_danger.get_long_message()
@pyqtProperty(QEAmount, notify=detailsChanged)
def amount(self):
@ -253,7 +253,7 @@ class QETxDetails(QObject, QtEventListener):
@pyqtProperty(bool, notify=detailsChanged)
def shouldConfirm(self):
return self._should_confirm
return self._sighash_danger.needs_confirm()
def update(self, from_txid: bool = False):
assert self._wallet
@ -299,8 +299,7 @@ class QETxDetails(QObject, QtEventListener):
fee_per_kb = txinfo.fee / size * 1000
self._feerate_str = self._wallet.wallet.config.format_fee_rate(fee_per_kb)
self._should_confirm = False
should_reject = False
self._sighash_danger = TxSighashDanger()
self._lock_delay = 0
self._in_mempool = False
@ -314,9 +313,7 @@ class QETxDetails(QObject, QtEventListener):
elif txinfo.tx_mined_status.height == TX_HEIGHT_FUTURE:
self._lock_delay = txinfo.tx_mined_status.wanted_height - self._wallet.wallet.adb.get_local_height()
if isinstance(self._tx, PartialTransaction):
self._should_confirm, should_reject, message = self._wallet.wallet.check_sighash(self._tx)
if message:
self._warning = '\n'.join([_('Danger! This transaction is non-standard!'), message])
self._sighash_danger = self._wallet.wallet.check_sighash(self._tx)
if self._wallet.wallet.lnworker:
# Calling lnworker.get_onchain_history and wallet.get_full_history here
@ -344,13 +341,11 @@ class QETxDetails(QObject, QtEventListener):
self._can_cpfp = txinfo.can_cpfp and not txinfo.can_remove
self._can_save_as_local = txinfo.can_save_as_local and not txinfo.can_remove
self._can_remove = txinfo.can_remove
self._can_sign = not self._is_complete and self._wallet.wallet.can_sign(self._tx) and not should_reject
if isinstance(self._tx, PartialTransaction):
risk_of_burning_coins = (self._can_sign and txinfo.fee is not None
and self._wallet.wallet.get_warning_for_risk_of_burning_coins_as_fees(self._tx))
if risk_of_burning_coins:
self._warning = '\n'.join([self._warning, risk_of_burning_coins])
self._can_sign = (
not self._is_complete
and self._wallet.wallet.can_sign(self._tx)
and not self._sighash_danger.needs_reject()
)
self.detailsChanged.emit()

41
electrum/gui/qt/transaction_dialog.py

@ -54,6 +54,7 @@ from electrum.transaction import SerializationError, Transaction, PartialTransac
from electrum.logging import get_logger
from electrum.util import ShortID, get_asyncio_loop
from electrum.network import Network
from electrum.wallet import TxSighashRiskLevel, TxSighashDanger
from . import util
from .util import (MessageBoxMixin, read_QIcon, Buttons, icon_path,
@ -115,9 +116,7 @@ class TxInOutWidget(QWidget):
self.sighash_label = QLabel()
self.sighash_label.setStyleSheet('font-weight: bold')
self.sighash_confirm = False
self.sighash_reject = False
self.sighash_message = ''
self.sighash_danger = TxSighashDanger()
self.inputs_warning_icon = QLabel()
pixmap = QPixmap(icon_path("warning"))
pixmap_size = round(2 * char_width_in_lineedit())
@ -262,11 +261,11 @@ class TxInOutWidget(QWidget):
)
if isinstance(self.tx, PartialTransaction):
self.sighash_confirm, self.sighash_reject, self.sighash_message = self.wallet.check_sighash(self.tx)
if self.sighash_message:
self.sighash_label.setText(_('Danger! This transaction is non-standard!'))
self.sighash_danger = self.wallet.check_sighash(self.tx)
if self.sighash_danger.risk_level >= TxSighashRiskLevel.WEIRD_SIGHASH:
self.sighash_label.setText(self.sighash_danger.short_message)
self.inputs_warning_icon.setVisible(True)
self.inputs_warning_icon.setToolTip(self.sighash_message)
self.inputs_warning_icon.setToolTip(self.sighash_danger.get_long_message())
self.outputs_header.setText(_("Outputs") + ' (%d)'%len(self.tx.outputs()))
o_text = self.outputs_textedit
@ -667,13 +666,15 @@ class TxDialog(QDialog, MessageBoxMixin):
self.update()
self.main_window.pop_top_level_window(self)
if self.io_widget.sighash_confirm:
if not self.question('\n'.join([
_('Danger! This transaction is non-standard!'),
self.io_widget.sighash_message,
'',
_('Are you sure you want to sign this transaction?')
])):
if self.io_widget.sighash_danger.needs_confirm():
if not self.question(
msg='\n'.join([
self.io_widget.sighash_danger.get_long_message(),
'',
_('Are you sure you want to sign this transaction?')
]),
title=self.io_widget.sighash_danger.short_message,
):
return
self.sign_button.setDisabled(True)
self.main_window.push_top_level_window(self)
@ -801,8 +802,9 @@ class TxDialog(QDialog, MessageBoxMixin):
self.broadcast_button.setEnabled(tx_details.can_broadcast)
can_sign = not self.tx.is_complete() and \
(self.wallet.can_sign(self.tx) or bool(self.external_keypairs))
self.sign_button.setEnabled(can_sign and not self.io_widget.sighash_reject)
self.sign_button.setToolTip(self.io_widget.sighash_message)
self.sign_button.setEnabled(can_sign and not self.io_widget.sighash_danger.needs_reject())
if sh_danger_msg := self.io_widget.sighash_danger.get_long_message():
self.sign_button.setToolTip(sh_danger_msg)
if tx_details.txid:
self.tx_hash_e.setText(tx_details.txid)
else:
@ -892,10 +894,9 @@ class TxDialog(QDialog, MessageBoxMixin):
color=ColorScheme.RED.as_color().name(),
)
if isinstance(self.tx, PartialTransaction):
risk_of_burning_coins = (can_sign and fee is not None
and self.wallet.get_warning_for_risk_of_burning_coins_as_fees(self.tx))
self.fee_warning_icon.setToolTip(str(risk_of_burning_coins))
self.fee_warning_icon.setVisible(bool(risk_of_burning_coins))
sh_warning = self.io_widget.sighash_danger.get_long_message()
self.fee_warning_icon.setToolTip(str(sh_warning))
self.fee_warning_icon.setVisible(can_sign and bool(sh_warning))
self.fee_label.setText(fee_str)
self.size_label.setText(size_str)
if ln_amount is None or ln_amount == 0:

7
electrum/tests/test_wallet_vertical.py

@ -13,7 +13,8 @@ from electrum import util
from electrum.address_synchronizer import TX_HEIGHT_UNCONFIRMED, TX_HEIGHT_UNCONF_PARENT
from electrum.wallet import (sweep, Multisig_Wallet, Standard_Wallet, Imported_Wallet,
restore_wallet_from_text, Abstract_Wallet, CannotBumpFee, BumpFeeStrategy,
TransactionPotentiallyDangerousException, TransactionDangerousException)
TransactionPotentiallyDangerousException, TransactionDangerousException,
TxSighashRiskLevel)
from electrum.util import bfh, NotEnoughFunds, UnrelatedTransactionException, UserFacingException
from electrum.transaction import Transaction, PartialTxOutput, tx_from_any, Sighash
from electrum.mnemonic import seed_type
@ -3018,6 +3019,7 @@ class TestWalletSending(ElectrumTestCase):
tx.inputs()[0].sighash = Sighash.NONE
tx.inputs()[1].sighash = Sighash.ALL
self.assertEqual(TxSighashRiskLevel.INSANE_SIGHASH, wallet1.check_sighash(tx).risk_level)
with self.assertRaises(TransactionDangerousException):
wallet1.sign_transaction(tx, password=None)
with self.assertRaises(TransactionDangerousException):
@ -3025,16 +3027,19 @@ class TestWalletSending(ElectrumTestCase):
tx.inputs()[0].sighash = Sighash.ALL
tx.inputs()[1].sighash = Sighash.SINGLE
self.assertEqual(TxSighashRiskLevel.WEIRD_SIGHASH, wallet1.check_sighash(tx).risk_level)
with self.assertRaises(TransactionPotentiallyDangerousException):
wallet1.sign_transaction(tx, password=None)
tx.inputs()[0].sighash = Sighash.ALL | Sighash.ANYONECANPAY
tx.inputs()[1].sighash = Sighash.ALL
self.assertEqual(TxSighashRiskLevel.WEIRD_SIGHASH, wallet1.check_sighash(tx).risk_level)
with self.assertRaises(TransactionPotentiallyDangerousException):
wallet1.sign_transaction(tx, password=None)
tx.inputs()[0].sighash = Sighash.ALL
tx.inputs()[1].sighash = Sighash.ALL
self.assertEqual(TxSighashRiskLevel.SAFE, wallet1.check_sighash(tx).risk_level)
self.assertFalse(tx.is_complete())
wallet1.sign_transaction(tx, password=None)
self.assertTrue(tx.is_complete())

180
electrum/wallet.py

@ -256,6 +256,60 @@ class TransactionPotentiallyDangerousException(Exception): pass
class TransactionDangerousException(TransactionPotentiallyDangerousException): pass
class TxSighashRiskLevel(enum.IntEnum):
# higher value -> more risk
SAFE = 0
FEE_WARNING_SKIPCONFIRM = 1 # show warning icon (ignored for CLI)
FEE_WARNING_NEEDCONFIRM = 2 # prompt user for confirmation
WEIRD_SIGHASH = 3 # prompt user for confirmation
INSANE_SIGHASH = 4 # reject
class TxSighashDanger:
def __init__(
self,
*,
risk_level: TxSighashRiskLevel = TxSighashRiskLevel.SAFE,
short_message: str = None,
messages: List[str] = None,
):
self.risk_level = risk_level
self.short_message = short_message
self._messages = messages or []
def needs_confirm(self) -> bool:
"""If True, the user should be prompted for explicit confirmation before signing."""
return self.risk_level >= TxSighashRiskLevel.FEE_WARNING_NEEDCONFIRM
def needs_reject(self) -> bool:
"""If True, the transaction should be rejected, i.e. abort signing."""
return self.risk_level >= TxSighashRiskLevel.INSANE_SIGHASH
def get_long_message(self) -> str:
"""Returns a description of the potential dangers of signing the tx that can be shown to the user.
Empty string if there are none.
"""
if self.short_message:
header = [self.short_message]
else:
header = []
return "\n".join(header + self._messages)
def combine(*args: 'TxSighashDanger') -> 'TxSighashDanger':
max_danger = max(args, key=lambda sighash_danger: sighash_danger.risk_level) # type: TxSighashDanger
messages = [msg for sighash_danger in args for msg in sighash_danger._messages]
return TxSighashDanger(
risk_level=max_danger.risk_level,
short_message=max_danger.short_message,
messages=messages,
)
def __repr__(self):
return (f"<{self.__class__.__name__} risk_level={self.risk_level} "
f"short_message={self.short_message!r} _messages={self._messages!r}>")
class BumpFeeStrategy(enum.Enum):
PRESERVE_PAYMENT = enum.auto()
DECREASE_PAYMENT = enum.auto()
@ -2478,21 +2532,11 @@ class Abstract_Wallet(ABC, Logger, EventListener):
return tx
# check if signing is dangerous
should_confirm, should_reject, message = self.check_sighash(tx)
if should_reject:
raise TransactionDangerousException('Not signing transaction:\n' + message)
if not ignore_warnings:
if should_confirm:
message = '\n'.join([_('Danger! This transaction is non-standard!'), message])
fee = self.get_wallet_delta(tx).fee
risk_of_burning_coins = (fee is not None
and self.get_warning_for_risk_of_burning_coins_as_fees(tx))
if risk_of_burning_coins:
should_confirm = True
message = '\n'.join([message, risk_of_burning_coins])
if should_confirm:
raise TransactionPotentiallyDangerousException('Not signing transaction:\n' + message)
sh_danger = self.check_sighash(tx)
if sh_danger.needs_reject():
raise TransactionDangerousException('Not signing transaction:\n' + sh_danger.get_long_message())
if sh_danger.needs_confirm() and not ignore_warnings:
raise TransactionPotentiallyDangerousException('Not signing transaction:\n' + sh_danger.get_long_message())
# add info to a temporary tx copy; including xpubs
# and full derivation paths as hw keystores might want them
@ -3037,8 +3081,8 @@ class Abstract_Wallet(ABC, Logger, EventListener):
self.sign_transaction(tx, password)
return tx
def get_warning_for_risk_of_burning_coins_as_fees(self, tx: 'PartialTransaction') -> Optional[str]:
"""Returns a warning message if there is risk of burning coins as fees if we sign.
def _check_risk_of_burning_coins_as_fees(self, tx: 'PartialTransaction') -> TxSighashDanger:
"""Helper method to check if there is risk of burning coins as fees if we sign.
Note that if not all inputs are ismine, e.g. coinjoin, the risk is not just about fees.
Note:
@ -3047,59 +3091,77 @@ class Abstract_Wallet(ABC, Logger, EventListener):
- BIP-taproot sighash commits to *all* input amounts
"""
assert isinstance(tx, PartialTransaction)
rl = TxSighashRiskLevel
short_message = _("Warning") + ": " + _("The fee could not be verified!")
# check that all inputs use SIGHASH_ALL
if not all(txin.sighash in (None, Sighash.ALL) for txin in tx.inputs()):
messages = [(_("Warning") + ": "
+ _("Some inputs use non-default sighash flags, which might affect the fee."))]
return TxSighashDanger(risk_level=rl.FEE_WARNING_NEEDCONFIRM, short_message=short_message, messages=messages)
# if we have all full previous txs, we *know* all the input amounts -> fine
if all([txin.utxo for txin in tx.inputs()]):
return None
return TxSighashDanger(risk_level=rl.SAFE)
# a single segwit input -> fine
if len(tx.inputs()) == 1 and tx.inputs()[0].is_segwit() and tx.inputs()[0].witness_utxo:
return None
return TxSighashDanger(risk_level=rl.SAFE)
# coinjoin or similar
if any([not self.is_mine(txin.address) for txin in tx.inputs()]):
return (_("Warning") + ": "
+ _("The input amounts could not be verified as the previous transactions are missing.\n"
"The amount of money being spent CANNOT be verified."))
messages = [(_("Warning") + ": "
+ _("The input amounts could not be verified as the previous transactions are missing.\n"
"The amount of money being spent CANNOT be verified."))]
return TxSighashDanger(risk_level=rl.FEE_WARNING_NEEDCONFIRM, short_message=short_message, messages=messages)
# some inputs are legacy
if any([not txin.is_segwit() for txin in tx.inputs()]):
return (_("Warning") + ": "
+ _("The fee could not be verified. Signing non-segwit inputs is risky:\n"
"if this transaction was maliciously modified before you sign,\n"
"you might end up paying a higher mining fee than displayed."))
messages = [(_("Warning") + ": "
+ _("The fee could not be verified. Signing non-segwit inputs is risky:\n"
"if this transaction was maliciously modified before you sign,\n"
"you might end up paying a higher mining fee than displayed."))]
return TxSighashDanger(risk_level=rl.FEE_WARNING_NEEDCONFIRM, short_message=short_message, messages=messages)
# all inputs are segwit
# https://lists.linuxfoundation.org/pipermail/bitcoin-dev/2017-August/014843.html
return (_("Warning") + ": "
+ _("If you received this transaction from an untrusted device, "
"do not accept to sign it more than once,\n"
"otherwise you could end up paying a different fee."))
def check_sighash(self, tx: 'PartialTransaction') -> (bool, bool, str):
"""Checks the Sighash for my inputs and return hints in the form
(confirm, reject, message), where confirm indicates the user should explicitly confirm,
reject indicates the transaction should be rejected (unless overruled in e.g. a config property),
and message containing a user-facing message describing the context"""
hintmap = dict([
(0, (False, False, None)),
(Sighash.NONE, (True, True, _('Input {} is marked SIGHASH_NONE.'))),
(Sighash.SINGLE, (True, False, _('Input {} is marked SIGHASH_SINGLE.'))),
(Sighash.ALL, (False, False, None)),
(Sighash.ANYONECANPAY, (True, False, _('Input {} is marked SIGHASH_ANYONECANPAY.')))
])
confirm = False
reject = False
messages = []
messages = [(_("Warning") + ": "
+ _("If you received this transaction from an untrusted device, "
"do not accept to sign it more than once,\n"
"otherwise you could end up paying a different fee."))]
return TxSighashDanger(risk_level=rl.FEE_WARNING_SKIPCONFIRM, short_message=short_message, messages=messages)
def check_sighash(self, tx: 'PartialTransaction') -> TxSighashDanger:
"""Checks the Sighash for my inputs and considers if the tx is safe to sign."""
assert isinstance(tx, PartialTransaction)
rl = TxSighashRiskLevel
hintmap = {
0: (rl.SAFE, None),
Sighash.NONE: (rl.INSANE_SIGHASH, _('Input {} is marked SIGHASH_NONE.')),
Sighash.SINGLE: (rl.WEIRD_SIGHASH, _('Input {} is marked SIGHASH_SINGLE.')),
Sighash.ALL: (rl.SAFE, None),
Sighash.ANYONECANPAY: (rl.WEIRD_SIGHASH, _('Input {} is marked SIGHASH_ANYONECANPAY.')),
}
sighash_danger = TxSighashDanger()
for txin_idx, txin in enumerate(tx.inputs()):
if txin.sighash and txin.sighash != Sighash.ALL: # non-standard
addr = self.adb.get_txin_address(txin)
if self.is_mine(addr):
sh_out = txin.sighash & (Sighash.ANYONECANPAY ^ 0xff)
sh_in = txin.sighash & Sighash.ANYONECANPAY
confirm |= hintmap[sh_out][0] | hintmap[sh_in][0]
reject |= hintmap[sh_out][1] | hintmap[sh_in][1]
for sh in [sh_out, sh_in]:
msg = hintmap[sh][2]
if msg:
messages.append('%s: %s' % (_('Fatal') if hintmap[sh][1] else _('Warning'),
msg.format(txin_idx)))
return confirm, reject, '\n'.join(messages)
if txin.sighash in (None, Sighash.ALL):
continue # None will get converted to Sighash.ALL, so these values are safe
# found interesting sighash flag
addr = self.adb.get_txin_address(txin)
if self.is_mine(addr):
sh_base = txin.sighash & (Sighash.ANYONECANPAY ^ 0xff)
sh_acp = txin.sighash & Sighash.ANYONECANPAY
for sh in [sh_base, sh_acp]:
if msg := hintmap[sh][1]:
risk_level = hintmap[sh][0]
header = _('Fatal') if TxSighashDanger(risk_level=risk_level).needs_reject() else _('Warning')
shd = TxSighashDanger(
risk_level=risk_level,
short_message=_('Danger! This transaction uses non-default sighash flags!'),
messages=[f"{header}: {msg.format(txin_idx)}"],
)
sighash_danger = sighash_danger.combine(shd)
if sighash_danger.needs_reject(): # no point for further tests
return sighash_danger
# if we show any fee to the user, check now how reliable that is:
if self.get_wallet_delta(tx).fee is not None:
shd = self._check_risk_of_burning_coins_as_fees(tx)
sighash_danger = sighash_danger.combine(shd)
return sighash_danger
def get_tx_fee_warning(
self, *,

Loading…
Cancel
Save