diff --git a/electrum/electrumabc_gui/qt/network_dialog.py b/electrum/electrumabc_gui/qt/network_dialog.py
index 300090473..29b86f73a 100644
--- a/electrum/electrumabc_gui/qt/network_dialog.py
+++ b/electrum/electrumabc_gui/qt/network_dialog.py
@@ -1,1467 +1,1471 @@
#!/usr/bin/env python3
#
# Electrum ABC - lightweight eCash client
# Copyright (C) 2020 The Electrum ABC developers
# Copyright (C) 2012 thomasv@gitorious
#
# Electron Cash - lightweight Bitcoin Cash client
# Copyright (C) 2020 The Electron Cash Developers
#
# Permission is hereby granted, free of charge, to any person
# obtaining a copy of this software and associated documentation files
# (the "Software"), to deal in the Software without restriction,
# including without limitation the rights to use, copy, modify, merge,
# publish, distribute, sublicense, and/or sell copies of the Software,
# and to permit persons to whom the Software is furnished to do so,
# subject to the following conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
# BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
# ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
# CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
import queue
import socket
from functools import partial
from PyQt5 import QtWidgets
from PyQt5.QtCore import QObject, Qt, QThread, QTimer, pyqtSignal
from PyQt5.QtGui import QIcon
from electrumabc import networks
from electrumabc.constants import PROJECT_NAME
from electrumabc.i18n import _, pgettext
from electrumabc.interface import Interface
from electrumabc.network import (
Network,
deserialize_server,
get_eligible_servers,
serialize_server,
)
from electrumabc.plugins import run_hook
from electrumabc.printerror import PrintError, print_error
from electrumabc.tor import TorController
from electrumabc.util import Weak, in_main_thread
from .tor_downloader import DownloadTorDialog
from .util import (
Buttons,
CloseButton,
HelpButton,
MessageBoxMixin,
PasswordLineEdit,
WindowModalDialog,
WWLabel,
char_width_in_lineedit,
rate_limited,
)
-from .utils import UserPortValidator
+from .utils.validators import HostValidator, PortValidator, UserPortValidator
protocol_names = ["TCP", "SSL"]
protocol_letters = "ts"
class NetworkDialog(MessageBoxMixin, QtWidgets.QDialog):
network_updated_signal = pyqtSignal()
def __init__(self, network: Network, config):
QtWidgets.QDialog.__init__(self)
self.setWindowTitle(_("Network"))
self.setMinimumSize(500, 350)
self.nlayout = NetworkChoiceLayout(self, network, config)
vbox = QtWidgets.QVBoxLayout(self)
vbox.addLayout(self.nlayout.layout())
# We don't want the close button's behavior to have the enter key close
# the window because user may edit text fields, etc, so we do the below:
close_but = CloseButton(self)
close_but.setDefault(False)
close_but.setAutoDefault(False)
vbox.addLayout(Buttons(close_but))
self.network_updated_signal.connect(self.on_update)
# below timer is to work around Qt on Linux display glitches when
# showing this window.
self.workaround_timer = QTimer()
self.workaround_timer.timeout.connect(self._workaround_update)
self.workaround_timer.setSingleShot(True)
network.register_callback(
self.on_network, ["blockchain_updated", "interfaces", "status"]
)
self.refresh_timer = QTimer(self)
self.refresh_timer.timeout.connect(self.network_updated_signal.emit)
self.refresh_timer.setInterval(500)
def jumpto(self, location: str):
self.nlayout.jumpto(location)
def on_network(self, event, *args):
"""This may run in network thread"""
# print_error("[NetworkDialog] on_network:",event,*args)
self.network_updated_signal.emit() # this enqueues call to on_update in GUI thread
@rate_limited(0.333)
def on_update(self):
"""This always runs in main GUI thread"""
self.nlayout.update()
def closeEvent(self, e):
# Warn if non-SSL mode when closing dialog
if (
not self.nlayout.ssl_cb.isChecked()
and not self.nlayout.tor_cb.isChecked()
and not self.nlayout.server_host.text().lower().endswith(".onion")
and not self.nlayout.config.get("non_ssl_noprompt", False)
):
ok, chk = self.question(
"".join(
[
_("You have selected non-SSL mode for your server settings."),
" ",
_("Using this mode presents a potential security risk."),
"\n\n",
_("Are you sure you wish to proceed?"),
]
),
detail_text="".join(
[
_(
"All of your traffic to the blockchain servers will be sent"
" unencrypted."
),
" ",
_(
"Additionally, you may also be vulnerable to"
" man-in-the-middle attacks."
),
" ",
_(
"It is strongly recommended that you go back and enable SSL"
" mode."
),
]
),
rich_text=False,
title=_("Security Warning"),
icon=QtWidgets.QMessageBox.Critical,
checkbox_text="Don't ask me again",
)
if chk:
self.nlayout.config.set_key("non_ssl_noprompt", True)
if not ok:
e.ignore()
return
super().closeEvent(e)
def hideEvent(self, e):
super().hideEvent(e)
if not self.isVisible():
self.workaround_timer.stop()
self.refresh_timer.stop()
def showEvent(self, e):
super().showEvent(e)
if e.isAccepted():
# Single-shot. Works around Linux/Qt bugs
# -- see _workaround_update below for description.
self.workaround_timer.start(500)
self.refresh_timer.start()
def _workaround_update(self):
# Hack to work around strange behavior on some Linux:
# On some Linux systems (Debian based), the dialog sometimes is empty
# and glitchy if we don't do this. Note this .update() call is a Qt
# C++ QWidget::update() call and has nothing to do with our own
# same-named `update` methods.
QtWidgets.QDialog.update(self)
class NodesListWidget(QtWidgets.QTreeWidget):
def __init__(self, parent):
QtWidgets.QTreeWidget.__init__(self)
self.parent = parent
self.setHeaderLabels([_("Connected node"), "", _("Height")])
self.setContextMenuPolicy(Qt.CustomContextMenu)
self.customContextMenuRequested.connect(self.create_menu)
def create_menu(self, position):
item = self.currentItem()
if not item:
return
is_server = not bool(item.data(0, Qt.UserRole))
menu = QtWidgets.QMenu()
if is_server:
server = item.data(1, Qt.UserRole)
menu.addAction(
_("Use as server"), lambda: self.parent.follow_server(server)
)
else:
index = item.data(1, Qt.UserRole)
menu.addAction(
_("Follow this branch"), lambda: self.parent.follow_branch(index)
)
menu.exec_(self.viewport().mapToGlobal(position))
def keyPressEvent(self, event):
if event.key() in {Qt.Key_F2, Qt.Key_Return}:
item, col = self.currentItem(), self.currentColumn()
if item and col > -1:
self.on_activated(item, col)
else:
super().keyPressEvent(event)
def on_activated(self, item, column):
# on 'enter' we show the menu
pt = self.visualItemRect(item).bottomLeft()
pt.setX(50)
self.customContextMenuRequested.emit(pt)
def update_servers(self, network, servers):
item = self.currentItem()
selection_data = None
if item:
selection_data = item.data(1, Qt.UserRole)
self.clear()
chains = network.get_blockchains()
n_chains = len(chains)
previous_server_item = None
for k, items in chains.items():
b = network.blockchains[k]
name = b.get_name()
if n_chains > 1:
# group the servers as children of a parent chain item
blockchain_root_item = QtWidgets.QTreeWidgetItem(
[name + f"@{b.get_base_height()}", "", f"{b.height()}"]
)
blockchain_root_item.setData(0, Qt.UserRole, 1)
blockchain_root_item.setData(1, Qt.UserRole, b.base_height)
self.addTopLevelItem(blockchain_root_item)
else:
# group servers as direct children of the tree widget
# (simple list)
blockchain_root_item = self.invisibleRootItem()
# Add servers
for i in items:
star = " ◀" if i == network.interface else ""
display_text = i.host
is_onion = i.host.lower().endswith(".onion")
if is_onion and i.host in servers and "display" in servers[i.host]:
display_text = servers[i.host]["display"] + " (.onion)"
item = QtWidgets.QTreeWidgetItem(
[display_text + star, "", "%d" % i.tip]
)
item.setData(0, Qt.UserRole, 0)
item.setData(1, Qt.UserRole, i.server)
if i.server == selection_data:
previous_server_item = item
if is_onion:
item.setIcon(1, QIcon(":icons/tor_logo.svg"))
blockchain_root_item.addChild(item)
blockchain_root_item.setExpanded(True)
# restore selection, if there was any
if previous_server_item:
self.setCurrentItem(previous_server_item)
h = self.header()
h.setStretchLastSection(False)
h.setSectionResizeMode(0, QtWidgets.QHeaderView.Stretch)
h.setSectionResizeMode(1, QtWidgets.QHeaderView.ResizeToContents)
h.setSectionResizeMode(2, QtWidgets.QHeaderView.ResizeToContents)
class ServerFlag:
"""Used by ServerListWidget for Server flags & Symbols"""
BadCertificate = 4 # Servers with a bad certificate.
Banned = 2 # Blacklisting/banning was a hidden mechanism inherited from Electrum. We would blacklist misbehaving servers under the hood. Now that facility is exposed (editable by the user). We never connect to blacklisted servers.
Preferred = 1 # Preferred servers (white-listed) start off as the servers in servers.json and are "more trusted" and optionally the user can elect to connect to only these servers
NoFlag = 0
Symbol = {NoFlag: "", Preferred: "⭐", Banned: "⛔", BadCertificate: "❗️"}
UnSymbol = { # used for "disable X" context menu
NoFlag: "",
Preferred: "❌",
Banned: "✅",
BadCertificate: "",
}
class ServerListWidget(QtWidgets.QTreeWidget):
def __init__(self, parent):
QtWidgets.QTreeWidget.__init__(self)
self.parent = parent
self.setHeaderLabels(["", _("Host"), "", _("Port")])
self.setContextMenuPolicy(Qt.CustomContextMenu)
self.customContextMenuRequested.connect(self.create_menu)
def create_menu(self, position):
item = self.currentItem()
if not item:
return
menu = QtWidgets.QMenu()
server = item.data(2, Qt.UserRole)
if self.parent.can_set_server(server):
useAction = menu.addAction(
_("Use as server"), lambda: self.set_server(server)
)
else:
useAction = menu.addAction(server.split(":", 1)[0], lambda: None)
useAction.setDisabled(True)
menu.addSeparator()
flagval = item.data(0, Qt.UserRole)
iswl = flagval & ServerFlag.Preferred
if flagval & ServerFlag.Banned:
optxt = ServerFlag.UnSymbol[ServerFlag.Banned] + " " + _("Unban server")
isbl = True
useAction.setDisabled(True)
useAction.setText(_("Server banned"))
else:
optxt = ServerFlag.Symbol[ServerFlag.Banned] + " " + _("Ban server")
isbl = False
if not isbl:
if flagval & ServerFlag.Preferred:
optxt_fav = (
ServerFlag.UnSymbol[ServerFlag.Preferred]
+ " "
+ _("Remove from preferred")
)
else:
optxt_fav = (
ServerFlag.Symbol[ServerFlag.Preferred]
+ " "
+ _("Add to preferred")
)
menu.addAction(
optxt_fav, lambda: self.parent.set_whitelisted(server, not iswl)
)
menu.addAction(optxt, lambda: self.parent.set_blacklisted(server, not isbl))
if flagval & ServerFlag.BadCertificate:
optxt = (
ServerFlag.UnSymbol[ServerFlag.BadCertificate]
+ " "
+ _("Remove pinned certificate")
)
menu.addAction(optxt, partial(self.on_remove_pinned_certificate, server))
menu.exec_(self.viewport().mapToGlobal(position))
def on_remove_pinned_certificate(self, server):
if not self.parent.remove_pinned_certificate(server):
QtWidgets.QMessageBox.critical(
None,
_("Remove pinned certificate"),
_("Failed to remove the pinned certificate. Check the log for errors."),
)
def set_server(self, s):
host, port, protocol = deserialize_server(s)
self.parent.server_host.setText(host)
self.parent.server_port.setText(port)
self.parent.autoconnect_cb.setChecked(
False
) # force auto-connect off if they did "Use as server"
self.parent.set_server()
self.parent.update()
def keyPressEvent(self, event):
if event.key() in [Qt.Key_F2, Qt.Key_Return]:
item, col = self.currentItem(), self.currentColumn()
if item and col > -1:
self.on_activated(item, col)
else:
super().keyPressEvent(event)
def on_activated(self, item, column):
# on 'enter' we show the menu
pt = self.visualItemRect(item).bottomLeft()
pt.setX(50)
self.customContextMenuRequested.emit(pt)
@staticmethod
def lightenItemText(item, rang=None):
if rang is None:
rang = range(0, item.columnCount())
for i in rang:
brush = item.foreground(i)
color = brush.color()
color.setHsvF(color.hueF(), color.saturationF(), 0.5)
brush.setColor(color)
item.setForeground(i, brush)
def update(self, network, servers, protocol, use_tor):
self.clear()
self.setIndentation(0)
wl_only = network.is_whitelist_only()
for _host, d in sorted(servers.items()):
is_onion = _host.lower().endswith(".onion")
if is_onion and not use_tor:
continue
port = d.get(protocol)
if port:
server = serialize_server(_host, port, protocol)
flag = ""
flagval = 0
tt = ""
if network.server_is_blacklisted(server):
flagval |= ServerFlag.Banned
if network.server_is_whitelisted(server):
flagval |= ServerFlag.Preferred
if network.server_is_bad_certificate(server):
flagval |= ServerFlag.BadCertificate
if flagval & ServerFlag.Banned:
flag = ServerFlag.Symbol[ServerFlag.Banned]
tt = _("This server is banned")
elif flagval & ServerFlag.BadCertificate:
flag = ServerFlag.Symbol[ServerFlag.BadCertificate]
tt = _(
"This server's pinned certificate mismatches its current"
" certificate"
)
elif flagval & ServerFlag.Preferred:
flag = ServerFlag.Symbol[ServerFlag.Preferred]
tt = _("This is a preferred server")
display_text = _host
if is_onion and "display" in d:
display_text = d["display"] + " (.onion)"
x = QtWidgets.QTreeWidgetItem([flag, display_text, "", port])
if is_onion:
x.setIcon(2, QIcon(":icons/tor_logo.svg"))
if tt:
x.setToolTip(0, tt)
if (
wl_only and not flagval & ServerFlag.Preferred
) or flagval & ServerFlag.Banned:
# lighten the text of servers we can't/won't connect to for the given mode
self.lightenItemText(x, range(1, 4))
x.setData(2, Qt.UserRole, server)
x.setData(0, Qt.UserRole, flagval)
x.setTextAlignment(0, Qt.AlignHCenter)
self.addTopLevelItem(x)
h = self.header()
h.setStretchLastSection(False)
h.setSectionResizeMode(0, QtWidgets.QHeaderView.ResizeToContents)
h.setSectionResizeMode(1, QtWidgets.QHeaderView.Stretch)
h.setSectionResizeMode(2, QtWidgets.QHeaderView.ResizeToContents)
h.setSectionResizeMode(3, QtWidgets.QHeaderView.ResizeToContents)
class NetworkChoiceLayout(QObject, PrintError):
def __init__(self, parent, network: Network, config, wizard=False):
super().__init__(parent)
self.network = network
self.config = config
self.protocol = None
self.tor_proxy = None
# tor detector
self.td = TorDetector(self, self.network)
self.td.found_proxy.connect(self.suggest_proxy)
self.tabs = tabs = QtWidgets.QTabWidget()
self.server_tab = server_tab = QtWidgets.QWidget()
weakTd = Weak.ref(self.td)
class ProxyTab(QtWidgets.QWidget):
def showEvent(slf, e):
super().showEvent(e)
td = weakTd()
if e.isAccepted() and td:
td.start() # starts the tor detector when proxy_tab appears
def hideEvent(slf, e):
super().hideEvent(e)
td = weakTd()
if e.isAccepted() and td:
td.stop() # stops the tor detector when proxy_tab disappears
self.proxy_tab = proxy_tab = ProxyTab()
self.blockchain_tab = blockchain_tab = QtWidgets.QWidget()
tabs.addTab(blockchain_tab, _("Overview"))
tabs.addTab(server_tab, _("Server"))
tabs.addTab(proxy_tab, _("Proxy"))
fixed_width_hostname = 24 * char_width_in_lineedit()
fixed_width_port = 6 * char_width_in_lineedit()
if wizard:
tabs.setCurrentIndex(1)
# server tab
grid = QtWidgets.QGridLayout(server_tab)
grid.setSpacing(8)
self.server_host = QtWidgets.QLineEdit()
self.server_host.setFixedWidth(fixed_width_hostname)
+ self.server_host.setValidator(HostValidator(self.server_host))
self.server_port = QtWidgets.QLineEdit()
self.server_port.setFixedWidth(fixed_width_port)
+ self.server_port.setValidator(PortValidator(self.server_port))
self.ssl_cb = QtWidgets.QCheckBox(_("Use SSL"))
self.autoconnect_cb = QtWidgets.QCheckBox(_("Select server automatically"))
self.autoconnect_cb.setEnabled(self.config.is_modifiable("auto_connect"))
weakSelf = Weak.ref(
self
) # Qt/Python GC hygiene: avoid strong references to self in lambda slots.
self.server_host.editingFinished.connect(
lambda: weakSelf() and weakSelf().set_server(onion_hack=True)
)
self.server_port.editingFinished.connect(
lambda: weakSelf() and weakSelf().set_server(onion_hack=True)
)
self.ssl_cb.clicked.connect(self.change_protocol)
self.autoconnect_cb.clicked.connect(self.set_server)
self.autoconnect_cb.clicked.connect(self.update)
msg = " ".join(
[
_(
f"If auto-connect is enabled, {PROJECT_NAME} will always use a "
"server that is on the longest blockchain."
),
_(
"If it is disabled, you have to choose a server you want to use."
f" {PROJECT_NAME} will warn you if your server is lagging."
),
]
)
grid.addWidget(self.autoconnect_cb, 0, 0, 1, 3)
grid.addWidget(HelpButton(msg), 0, 4)
self.preferred_only_cb = QtWidgets.QCheckBox(
_("Connect only to preferred servers")
)
self.preferred_only_cb.setEnabled(
self.config.is_modifiable("whitelist_servers_only")
)
self.preferred_only_cb.setToolTip(
_(
f"If enabled, restricts {PROJECT_NAME} to connecting to "
"servers only marked as 'preferred'."
)
)
self.preferred_only_cb.clicked.connect(
self.set_whitelisted_only
) # re-set the config key and notify network.py
msg = "\n\n".join(
[
_(
"If 'Connect only to preferred servers' is enabled, "
f"{PROJECT_NAME} will only connect to servers marked as "
f"'preferred' servers ({ServerFlag.Symbol[ServerFlag.Preferred]})."
),
_(
"This feature was added in response to the potential for a "
"malicious actor to deny service via launching many servers "
"(aka a sybil attack)."
),
_(
"If unsure, most of the time it's safe to leave this option "
"disabled. However leaving it enabled is safer (if a little "
"bit discouraging to new server operators wanting to populate "
"their servers)."
),
]
)
grid.addWidget(self.preferred_only_cb, 1, 0, 1, 3)
grid.addWidget(HelpButton(msg), 1, 4)
grid.addWidget(self.ssl_cb, 2, 0, 1, 3)
self.ssl_help = HelpButton(
_(
"SSL is used to authenticate and encrypt your connections with the"
" blockchain servers."
)
+ "\n\n"
+ _(
"Due to potential security risks, you may only disable SSL when using a"
" Tor Proxy."
)
)
grid.addWidget(self.ssl_help, 2, 4)
grid.addWidget(QtWidgets.QLabel(_("Server") + ":"), 3, 0)
grid.addWidget(self.server_host, 3, 1, 1, 2)
grid.addWidget(self.server_port, 3, 3)
# will get set by self.update()
self.server_list_label = label = QtWidgets.QLabel("")
grid.addWidget(label, 4, 0, 1, 5)
self.servers_list = ServerListWidget(self)
grid.addWidget(self.servers_list, 5, 0, 1, 5)
# will get populated with the legend by self.update()
self.legend_label = label = WWLabel("")
label.setTextInteractionFlags(
label.textInteractionFlags() & (~Qt.TextSelectableByMouse)
) # disable text selection by mouse here
self.legend_label.linkActivated.connect(self.on_view_blacklist)
grid.addWidget(label, 6, 0, 1, 4)
msg = " ".join(
[
_(
"Preferred servers ({}) are servers you have designated as reliable"
" and/or trustworthy."
).format(ServerFlag.Symbol[ServerFlag.Preferred]),
_(
"Initially, the preferred list is the hard-coded list of "
f"known-good servers vetted by the {PROJECT_NAME} developers."
),
_(
"You can add or remove any server from this list and"
" optionally elect to only connect to preferred servers."
),
"\n\n"
+ _(
f"Banned servers ({ServerFlag.Symbol[ServerFlag.Banned]}) "
"are servers deemed unreliable and/or untrustworthy, and "
f"so they will never be connected-to by {PROJECT_NAME}"
),
]
)
grid.addWidget(HelpButton(msg), 6, 4)
# Proxy tab
grid = QtWidgets.QGridLayout(proxy_tab)
grid.setSpacing(8)
# proxy setting
self.proxy_cb = QtWidgets.QCheckBox(_("Use proxy"))
self.proxy_cb.setToolTip(
_(
"If enabled, all connections application-wide will be routed through"
" this proxy."
)
)
self.proxy_cb.clicked.connect(self.check_disable_proxy)
self.proxy_cb.clicked.connect(self.set_proxy)
self.proxy_mode = QtWidgets.QComboBox()
self.proxy_mode.addItems(["SOCKS4", "SOCKS5", "HTTP"])
self.proxy_host = QtWidgets.QLineEdit()
self.proxy_host.setFixedWidth(fixed_width_hostname)
+ self.proxy_host.setValidator(HostValidator(self.proxy_host))
self.proxy_port = QtWidgets.QLineEdit()
self.proxy_port.setFixedWidth(fixed_width_port)
+ self.proxy_port.setValidator(PortValidator(self.proxy_port))
self.proxy_user = QtWidgets.QLineEdit()
self.proxy_user.setPlaceholderText(_("Proxy user"))
self.proxy_password = PasswordLineEdit()
self.proxy_password.setPlaceholderText(_("Password"))
self.proxy_mode.currentIndexChanged.connect(self.set_proxy)
self.proxy_host.editingFinished.connect(self.set_proxy)
self.proxy_port.editingFinished.connect(self.set_proxy)
self.proxy_user.editingFinished.connect(self.set_proxy)
self.proxy_password.editingFinished.connect(self.set_proxy)
self.tor_cb = QtWidgets.QCheckBox(_("Use Tor Proxy"))
self.tor_cb.setIcon(QIcon(":icons/tor_logo.svg"))
self.tor_cb.setEnabled(False)
self.tor_cb.clicked.connect(self.use_tor_proxy)
tor_proxy_tooltip = _(
"If enabled, all connections application-wide will be routed through Tor."
)
tor_proxy_help = (
tor_proxy_tooltip
+ "\n\n"
+ _(
"Depending on your configuration and preferences as a user, this may or"
" may not be ideal. In general, connections routed through Tor hide"
" your IP address from servers, at the expense of performance and"
" network throughput."
)
+ "\n\n"
+ _(
"For the average user, it's recommended that you leave this option "
"disabled and only leave the 'Start Tor client' option enabled."
)
)
self.tor_cb.setToolTip(tor_proxy_tooltip)
self.tor_enabled = QtWidgets.QCheckBox()
self.tor_enabled.setIcon(QIcon(":icons/tor_logo.svg"))
self.tor_enabled.clicked.connect(self.set_tor_enabled)
self.tor_enabled.setChecked(self.network.tor_controller.is_enabled())
self.tor_enabled_help = HelpButton("")
self.tor_custom_port_cb = QtWidgets.QCheckBox(_("Custom port"))
self.tor_enabled.clicked.connect(self.tor_custom_port_cb.setEnabled)
self.tor_custom_port_cb.setChecked(
bool(self.network.tor_controller.get_socks_port())
)
self.tor_custom_port_cb.clicked.connect(self.on_custom_port_cb_click)
custom_port_tooltip = _("Leave unspecified to automatically allocate a port.")
self.tor_custom_port_cb.setToolTip(custom_port_tooltip)
self.network.tor_controller.status_changed.append_weak(
self.on_tor_status_changed
)
self.tor_socks_port = QtWidgets.QLineEdit()
self.tor_socks_port.setFixedWidth(fixed_width_port)
self.tor_socks_port.editingFinished.connect(self.set_tor_socks_port)
self.tor_socks_port.setText(str(self.network.tor_controller.get_socks_port()))
self.tor_socks_port.setToolTip(custom_port_tooltip)
self.tor_socks_port.setValidator(
UserPortValidator(self.tor_socks_port, accept_zero=True)
)
self.dl_tor_button = QtWidgets.QPushButton(_("Download Tor"))
self.dl_tor_button.clicked.connect(self._show_download_tor_dialog)
self.dl_tor_help = HelpButton(
_("Download a compiled Tor executable.")
+ "\n\n"
+ _(
f"A Tor executable is provided by {PROJECT_NAME} for convenience, if "
"you don't know how to install Tor on your operating system. Linux "
"users are advised to install Tor via their package manager instead."
)
)
self.update_tor_enabled()
# Start Tor
grid.addWidget(self.dl_tor_button, 0, 1, 1, 2)
grid.addWidget(self.dl_tor_help, 0, 4)
grid.addWidget(self.tor_enabled, 1, 0, 1, 2)
grid.addWidget(self.tor_enabled_help, 1, 4)
# Custom Tor port
hbox = QtWidgets.QHBoxLayout()
hbox.addSpacing(20) # indentation
hbox.addWidget(self.tor_custom_port_cb, 0, Qt.AlignLeft | Qt.AlignVCenter)
hbox.addWidget(self.tor_socks_port, 0, Qt.AlignLeft | Qt.AlignVCenter)
hbox.addStretch(2)
hbox.setContentsMargins(0, 0, 0, 6) # a bit of a "paragraph break" here
grid.addLayout(hbox, 2, 0, 1, 3)
grid.addWidget(HelpButton(custom_port_tooltip), 2, 4)
# Use Tor Proxy
grid.addWidget(self.tor_cb, 3, 0, 1, 3)
grid.addWidget(HelpButton(tor_proxy_help), 3, 4)
# Proxy settings
grid.addWidget(self.proxy_cb, 4, 0, 1, 3)
grid.addWidget(
HelpButton(
_(
f"Proxy settings apply to all connections: with {PROJECT_NAME}"
" servers, but also with third-party services."
)
),
4,
4,
)
grid.addWidget(self.proxy_mode, 6, 1)
grid.addWidget(self.proxy_host, 6, 2)
grid.addWidget(self.proxy_port, 6, 3)
sublayout = QtWidgets.QHBoxLayout()
sublayout.addWidget(self.proxy_user)
sublayout.addWidget(self.proxy_password)
grid.addLayout(sublayout, 7, 2, 1, 2)
grid.setRowStretch(8, 1)
# Blockchain Tab
grid = QtWidgets.QGridLayout(blockchain_tab)
msg = " ".join(
[
_(
f"{PROJECT_NAME} connects to several nodes in order to "
"download block headers and find out the longest blockchain."
),
_(
"This blockchain is used to verify the transactions sent by "
"your transaction server."
),
]
)
row = 0
self.status_label = QtWidgets.QLabel("")
self.status_label.setTextInteractionFlags(
self.status_label.textInteractionFlags() | Qt.TextSelectableByMouse
)
grid.addWidget(QtWidgets.QLabel(_("Status") + ":"), row, 0)
grid.addWidget(self.status_label, row, 1, 1, 3)
grid.addWidget(HelpButton(msg), row, 4)
row += 1
self.server_label = QtWidgets.QLabel("")
self.server_label.setTextInteractionFlags(
self.server_label.textInteractionFlags() | Qt.TextSelectableByMouse
)
msg = _(
f"{PROJECT_NAME} sends your wallet addresses to a single "
"server, in order to receive your transaction history."
)
grid.addWidget(QtWidgets.QLabel(_("Server") + ":"), row, 0)
grid.addWidget(self.server_label, row, 1, 1, 3)
grid.addWidget(HelpButton(msg), row, 4)
row += 1
self.height_label = QtWidgets.QLabel("")
self.height_label.setTextInteractionFlags(
self.height_label.textInteractionFlags() | Qt.TextSelectableByMouse
)
msg = _("This is the height of your local copy of the blockchain.")
grid.addWidget(QtWidgets.QLabel(_("Blockchain") + ":"), row, 0)
grid.addWidget(self.height_label, row, 1)
grid.addWidget(HelpButton(msg), row, 4)
row += 1
self.reqs_label = QtWidgets.QLabel("")
self.reqs_label.setTextInteractionFlags(
self.height_label.textInteractionFlags() | Qt.TextSelectableByMouse
)
msg = _(
"The number of unanswered network requests.\n\n"
"You can configure:\n\n"
" - Limit: maximum request backlog size\n"
" - ChunkSize: requests to enqueue every 100ms\n\n"
"If the connection drops when synchronizing, you may wish "
"to reduce these values to throttle requests to the server."
)
grid.addWidget(QtWidgets.QLabel(_("Pending requests") + ":"), row, 0)
hbox = QtWidgets.QHBoxLayout()
hbox.addWidget(self.reqs_label)
hbox.setContentsMargins(0, 0, 12, 0)
hbox.addWidget(QtWidgets.QLabel(_("Limit:")))
self.req_max_sb = sb = QtWidgets.QSpinBox()
sb.setRange(1, 2000)
sb.setFocusPolicy(Qt.TabFocus | Qt.ClickFocus | Qt.WheelFocus)
hbox.addWidget(sb)
hbox.addWidget(QtWidgets.QLabel(_("ChunkSize:")))
self.req_chunk_sb = sb = QtWidgets.QSpinBox()
sb.setRange(1, 100)
sb.setFocusPolicy(Qt.TabFocus | Qt.ClickFocus | Qt.WheelFocus)
hbox.addWidget(sb)
but = QtWidgets.QPushButton(_("Reset"))
f = but.font()
f.setPointSize(f.pointSize() - 2)
but.setFont(f)
but.setDefault(False)
but.setAutoDefault(False)
hbox.addWidget(but)
grid.addLayout(hbox, row, 1, 1, 3)
grid.setAlignment(hbox, Qt.AlignLeft | Qt.AlignVCenter)
grid.setColumnStretch(3, 1)
grid.addWidget(HelpButton(msg), row, 4)
row += 1
def req_max_changed(val):
Interface.set_req_throttle_params(self.config, max_unanswered_requests=val)
def req_chunk_changed(val):
Interface.set_req_throttle_params(self.config, chunkSize=val)
def req_defaults():
p = Interface.req_throttle_default
Interface.set_req_throttle_params(
self.config, max_unanswered_requests=p.max, chunkSize=p.chunkSize
)
self.update()
but.clicked.connect(req_defaults)
self.req_max_sb.valueChanged.connect(req_max_changed)
self.req_chunk_sb.valueChanged.connect(req_chunk_changed)
self.split_label = QtWidgets.QLabel("")
self.split_label.setTextInteractionFlags(
self.split_label.textInteractionFlags() | Qt.TextSelectableByMouse
)
grid.addWidget(self.split_label, row, 0, 1, 3)
row += 2
self.nodes_list_widget = NodesListWidget(self)
grid.addWidget(self.nodes_list_widget, row, 0, 1, 5)
row += 1
vbox = QtWidgets.QVBoxLayout()
vbox.addWidget(tabs)
self.layout_ = vbox
self.network.tor_controller.active_port_changed.append_weak(
self.on_tor_port_changed
)
self.network.server_list_updated.append_weak(self.on_server_list_updated)
self.fill_in_proxy_settings()
self.update()
_tor_client_names = {
TorController.BinaryType.MISSING: _("Tor"),
TorController.BinaryType.SYSTEM: _("system Tor"),
TorController.BinaryType.DOWNLOADED: _("downloaded Tor"),
}
def update_tor_enabled(self, *args):
tbt = self.network.tor_controller.tor_binary_type
tbname = self._tor_client_names[tbt]
available = tbt != TorController.BinaryType.MISSING
self.dl_tor_button.setVisible(not available)
self.dl_tor_help.setVisible(not available)
self.tor_enabled.setText(
_("Start {tor_binary_name} client").format(
tor_binary_name=tbname,
tor_binary_name_capitalized=tbname[:1].upper() + tbname[1:],
)
)
self.tor_enabled.setEnabled(available)
self.tor_custom_port_cb.setEnabled(available and self.tor_enabled.isChecked())
self.tor_socks_port.setEnabled(
available and self.tor_custom_port_cb.isChecked()
)
tor_enabled_tooltip = [
_(
"This will start a private instance of the Tor proxy "
f"controlled by {PROJECT_NAME}."
)
]
if not available:
tor_enabled_tooltip.insert(
0, _("This feature is unavailable because no Tor binary was found.")
)
tor_enabled_tooltip_text = " ".join(tor_enabled_tooltip)
self.tor_enabled.setToolTip(tor_enabled_tooltip_text)
self.tor_enabled_help.help_text = (
tor_enabled_tooltip_text
+ "\n\n"
+ _(
"If unsure, it's safe to enable this feature, and leave 'Use Tor Proxy'"
" disabled. In that situation, only certain plugins (such as"
" CashFusion) will use Tor, but your regular SPV server connections"
" will remain unaffected."
)
)
def jumpto(self, location: str):
if not isinstance(location, str):
return
location = location.strip().lower()
if location in ("proxy", "tor"):
self.tabs.setCurrentWidget(self.proxy_tab)
elif location in ("servers", "server"):
self.tabs.setCurrentWidget(self.server_tab)
elif location in ("blockchain", "overview", "main"):
self.tabs.setCurrentWidget(self.blockchain_tab)
elif not run_hook("on_network_dialog_jumpto", self, location):
self.print_error(f"jumpto: unknown location '{location}'")
@in_main_thread
def on_tor_port_changed(self, controller: TorController):
if (
not controller.active_socks_port
or not controller.is_enabled()
or not self.tor_use
):
return
# The Network class handles actually changing the port, we just
# set the value in the text box here.
self.proxy_port.setText(str(controller.active_socks_port))
@in_main_thread
def on_server_list_updated(self):
self.update()
def check_disable_proxy(self, b):
if not self.config.is_modifiable("proxy"):
b = False
if self.tor_use:
# Disallow changing the proxy settings when Tor is in use
b = False
for w in [
self.proxy_mode,
self.proxy_host,
self.proxy_port,
self.proxy_user,
self.proxy_password,
]:
w.setEnabled(b)
def get_set_server_flags(self):
return (
self.config.is_modifiable("server"),
(
not self.autoconnect_cb.isChecked()
and not self.preferred_only_cb.isChecked()
),
)
def can_set_server(self, server):
return bool(
self.get_set_server_flags()[0]
and not self.network.server_is_blacklisted(server)
and (
not self.network.is_whitelist_only()
or self.network.server_is_whitelisted(server)
)
)
def enable_set_server(self):
modifiable, notauto = self.get_set_server_flags()
if modifiable:
self.server_host.setEnabled(notauto)
self.server_port.setEnabled(notauto)
else:
for w in [self.autoconnect_cb, self.server_host, self.server_port]:
w.setEnabled(False)
def update(self):
host, port, protocol, proxy_config, auto_connect = self.network.get_parameters()
preferred_only = self.network.is_whitelist_only()
if not self.server_host.hasFocus() and not self.server_port.hasFocus():
self.server_host.setText(host)
self.server_port.setText(port)
self.ssl_cb.setChecked(protocol == "s")
ssl_disable = (
self.ssl_cb.isChecked()
and not self.tor_cb.isChecked()
and not host.lower().endswith(".onion")
)
for w in [self.ssl_cb]: # , self.ssl_help]:
w.setDisabled(ssl_disable)
self.autoconnect_cb.setChecked(auto_connect)
self.preferred_only_cb.setChecked(preferred_only)
self.servers = self.network.get_servers()
host = (
self.network.interface.host
if self.network.interface
else pgettext("Referencing server", "None")
)
is_onion = host.lower().endswith(".onion")
if is_onion and host in self.servers and "display" in self.servers[host]:
host = self.servers[host]["display"] + " (.onion)"
self.server_label.setText(host)
self.set_protocol(protocol)
def protocol_suffix():
if protocol == "t":
return " (non-SSL)"
elif protocol == "s":
return " [SSL]"
return ""
server_list_txt = (
_("Server peers") if self.network.is_connected() else _("Servers")
) + " ({})".format(len(self.servers))
server_list_txt += protocol_suffix()
self.server_list_label.setText(server_list_txt)
if self.network.blacklisted_servers:
bl_srv_ct_str = ' ({}) {}'.format(
len(self.network.blacklisted_servers), _("View ban list...")
)
else:
bl_srv_ct_str = " (0) " # ensure rich text
servers_whitelisted = (
set(get_eligible_servers(self.servers, protocol)).intersection(
self.network.whitelisted_servers
)
- self.network.blacklisted_servers
)
self.legend_label.setText(
ServerFlag.Symbol[ServerFlag.Preferred]
+ "="
+ _("Preferred")
+ " ({})".format(len(servers_whitelisted))
+ " "
+ ServerFlag.Symbol[ServerFlag.Banned]
+ "="
+ _("Banned")
+ bl_srv_ct_str
)
self.servers_list.update(
self.network, self.servers, self.protocol, self.tor_cb.isChecked()
)
self.enable_set_server()
height_str = "%d " % (self.network.get_local_height()) + _("blocks")
self.height_label.setText(height_str)
n = len(self.network.get_interfaces())
status = _("Connected to %d nodes.") % n if n else _("Not connected")
if n:
status += protocol_suffix()
self.status_label.setText(status)
chains = self.network.get_blockchains()
if len(chains) > 1:
chain = self.network.blockchain()
checkpoint = chain.get_base_height()
name = chain.get_name()
msg = _("Chain split detected at block %d") % checkpoint + "\n"
msg += (
(
_("You are following branch")
if auto_connect
else _("Your server is on branch")
)
+ " "
+ name
)
msg += " (%d %s)" % (chain.get_branch_size(), _("blocks"))
else:
msg = ""
self.split_label.setText(msg)
self.reqs_label.setText(
str(
(
self.network.interface
and len(self.network.interface.unanswered_requests)
)
or 0
)
)
params = Interface.get_req_throttle_params(self.config)
self.req_max_sb.setValue(params.max)
self.req_chunk_sb.setValue(params.chunkSize)
self.nodes_list_widget.update_servers(self.network, self.servers)
def fill_in_proxy_settings(self):
host, port, protocol, proxy_config, auto_connect = self.network.get_parameters()
if not proxy_config:
proxy_config = {"mode": "none", "host": "localhost", "port": "9050"}
# We need to restore the "Use tor" checkbox as its value is needed in the server
# list, to determine whether to show .onion servers, before the TorDetector
# has been started.
self._set_tor_use(self.config.get("tor_use", False))
b = proxy_config.get("mode") != "none"
self.check_disable_proxy(b)
if b:
self.proxy_cb.setChecked(True)
self.proxy_mode.setCurrentIndex(
self.proxy_mode.findText(str(proxy_config.get("mode").upper()))
)
self.proxy_host.setText(proxy_config.get("host"))
self.proxy_port.setText(proxy_config.get("port"))
self.proxy_user.setText(proxy_config.get("user", ""))
self.proxy_password.setText(proxy_config.get("password", ""))
def layout(self):
return self.layout_
def set_protocol(self, protocol):
if protocol != self.protocol:
self.protocol = protocol
def change_protocol(self, use_ssl):
p = "s" if use_ssl else "t"
host = self.server_host.text()
pp = self.servers.get(host, networks.net.DEFAULT_PORTS)
if p not in pp.keys():
p = list(pp.keys())[0]
port = pp[p]
self.server_host.setText(host)
self.server_port.setText(port)
self.set_protocol(p)
self.set_server()
def follow_branch(self, index):
self.network.follow_chain(index)
self.update()
def follow_server(self, server):
self.network.switch_to_interface(server)
host, port, protocol, proxy, auto_connect = self.network.get_parameters()
host, port, protocol = deserialize_server(server)
self.network.set_parameters(host, port, protocol, proxy, auto_connect)
self.update()
def server_changed(self, x):
if x:
self.change_server(str(x.text(0)), self.protocol)
def change_server(self, host, protocol):
pp = self.servers.get(host, networks.net.DEFAULT_PORTS)
if protocol and protocol not in protocol_letters:
protocol = None
if protocol:
port = pp.get(protocol)
if port is None:
protocol = None
if not protocol:
if "s" in pp.keys():
protocol = "s"
port = pp.get(protocol)
else:
protocol = list(pp.keys())[0]
port = pp.get(protocol)
self.server_host.setText(host)
self.server_port.setText(port)
self.ssl_cb.setChecked(protocol == "s")
def accept(self):
pass
def set_server(self, onion_hack=False):
host, port, protocol, proxy, auto_connect = self.network.get_parameters()
host = str(self.server_host.text()).strip()
port = str(self.server_port.text()).strip()
protocol = "s" if self.ssl_cb.isChecked() else "t"
if onion_hack:
# Fix #1174 -- bring back from the dead non-SSL support for .onion only in a safe way
if host.lower().endswith(".onion"):
self.print_error(
"Onion/TCP hack: detected .onion, forcing TCP (non-SSL) mode"
)
protocol = "t"
self.ssl_cb.setChecked(False)
auto_connect = self.autoconnect_cb.isChecked()
self.network.set_parameters(host, port, protocol, proxy, auto_connect)
def set_proxy(self):
host, port, protocol, proxy, auto_connect = self.network.get_parameters()
if self.proxy_cb.isChecked():
proxy = {
"mode": str(self.proxy_mode.currentText()).lower(),
"host": str(self.proxy_host.text()),
"port": str(self.proxy_port.text()),
"user": str(self.proxy_user.text()),
"password": str(self.proxy_password.text()),
}
else:
proxy = None
self.network.set_parameters(host, port, protocol, proxy, auto_connect)
def suggest_proxy(self, found_proxy):
if not found_proxy:
self.tor_cb.setEnabled(False)
# It's not clear to me that if the tor service goes away and comes back
# later, and in the meantime they unchecked proxy_cb, that this should
# remain checked. I can see it being confusing for that to be the case.
# Better to uncheck. It gets auto-re-checked anyway if it comes back and
# it's the same due to code below. -Calin
self._set_tor_use(False)
return
self.tor_proxy = found_proxy
self.tor_cb.setText(
_("Use Tor proxy at port {tor_port}").format(tor_port=found_proxy[1])
)
same_proxy = (
self.proxy_mode.currentIndex() == self.proxy_mode.findText("SOCKS5")
and self.proxy_host.text() == found_proxy[0]
and self.proxy_port.text() == str(found_proxy[1])
and self.proxy_cb.isChecked()
)
self._set_tor_use(same_proxy)
self.tor_cb.setEnabled(True)
def _set_tor_use(self, use_it):
self.tor_use = use_it
self.config.set_key("tor_use", self.tor_use)
self.tor_cb.setChecked(self.tor_use)
self.proxy_cb.setEnabled(not self.tor_use)
self.check_disable_proxy(not self.tor_use)
def use_tor_proxy(self, use_it):
self._set_tor_use(use_it)
if not use_it:
self.proxy_cb.setChecked(False)
else:
socks5_mode_index = self.proxy_mode.findText("SOCKS5")
if socks5_mode_index == -1:
print_error("[network_dialog] can't find proxy_mode 'SOCKS5'")
return
self.proxy_mode.setCurrentIndex(socks5_mode_index)
self.proxy_host.setText("127.0.0.1")
self.proxy_port.setText(str(self.tor_proxy[1]))
self.proxy_user.setText("")
self.proxy_password.setText("")
self.proxy_cb.setChecked(True)
self.set_proxy()
def set_tor_enabled(self, enabled: bool):
self.network.tor_controller.set_enabled(enabled)
@in_main_thread
def on_tor_status_changed(self, controller):
if controller.status == TorController.Status.ERRORED and self.tabs.isVisible():
tbname = self._tor_client_names[self.network.tor_controller.tor_binary_type]
msg = _(
"The {tor_binary_name} client experienced an error or could not be"
" started."
).format(tor_binary_name=tbname)
QtWidgets.QMessageBox.critical(None, _("Tor Client Error"), msg)
def set_tor_socks_port(self):
socks_port = int(self.tor_socks_port.text())
self.network.tor_controller.set_socks_port(socks_port)
def on_custom_port_cb_click(self, b):
self.tor_socks_port.setEnabled(b)
if not b:
self.tor_socks_port.setText("0")
self.set_tor_socks_port()
def proxy_settings_changed(self):
self.tor_cb.setChecked(False)
def remove_pinned_certificate(self, server):
return self.network.remove_pinned_certificate(server)
def set_blacklisted(self, server, bl):
self.network.server_set_blacklisted(server, bl, True)
# if the blacklisted server is the active server, this will force a reconnect
# to another server
self.set_server()
self.update()
def set_whitelisted(self, server, flag):
self.network.server_set_whitelisted(server, flag, True)
self.set_server()
self.update()
def set_whitelisted_only(self, b):
self.network.set_whitelist_only(b)
# forces us to send a set-server to network.py which recomputes eligible
# servers, etc
self.set_server()
self.update()
def on_view_blacklist(self, ignored):
"""The 'view ban list...' link leads to a modal dialog box where the
user has the option to clear the entire blacklist. Build that dialog here."""
bl = sorted(self.network.blacklisted_servers)
parent = self.parent()
if not bl:
parent.show_error(_("Server ban list is empty!"))
return
d = WindowModalDialog(parent.top_level_window(), _("Banned Servers"))
vbox = QtWidgets.QVBoxLayout(d)
vbox.addWidget(QtWidgets.QLabel(_("Banned Servers") + " ({})".format(len(bl))))
tree = QtWidgets.QTreeWidget()
tree.setHeaderLabels([_("Host"), _("Port")])
for s in bl:
host, port, protocol = deserialize_server(s)
item = QtWidgets.QTreeWidgetItem([host, str(port)])
item.setFlags(Qt.ItemIsEnabled)
tree.addTopLevelItem(item)
tree.setIndentation(3)
h = tree.header()
h.setStretchLastSection(False)
h.setSectionResizeMode(0, QtWidgets.QHeaderView.Stretch)
h.setSectionResizeMode(1, QtWidgets.QHeaderView.ResizeToContents)
vbox.addWidget(tree)
clear_but = QtWidgets.QPushButton(_("Clear ban list"))
weakSelf = Weak.ref(self)
weakD = Weak.ref(d)
clear_but.clicked.connect(
lambda: weakSelf() and weakSelf().on_clear_blacklist() and weakD().reject()
)
vbox.addLayout(Buttons(clear_but, CloseButton(d)))
d.exec_()
def on_clear_blacklist(self):
bl = list(self.network.blacklisted_servers)
blen = len(bl)
if self.parent().question(
_("Clear all {} servers from the ban list?").format(blen)
):
for i, s in enumerate(bl):
self.network.server_set_blacklisted(
s, False, save=bool(i + 1 == blen)
) # save on last iter
self.update()
return True
return False
def _show_download_tor_dialog(self):
dialog = DownloadTorDialog(self.config, self.parent())
dialog.exec_()
# Let TorController know about the new binary
if dialog.was_download_successful:
self.network.tor_controller.detect_tor()
class TorDetector(QThread):
found_proxy = pyqtSignal(object)
def __init__(self, parent, network: Network):
super().__init__(parent)
self.network = network
self.network.tor_controller.active_port_changed.append_weak(
self.on_tor_port_changed
)
def on_tor_port_changed(self, controller: TorController):
if controller.active_socks_port and self.isRunning():
self.stopQ.put("kick")
def start(self):
# create a new stopQ blowing away the old one just in case it has old data in
# it (this prevents races with stop/start arriving too quickly for the thread)
self.stopQ = queue.Queue()
super().start()
def stop(self):
if self.isRunning():
self.stopQ.put(None)
self.wait()
def run(self):
while True:
ports = [9050, 9150] # Probable ports for Tor to listen at
if (
self.network.tor_controller
and self.network.tor_controller.is_enabled()
and self.network.tor_controller.active_socks_port
):
ports.insert(0, self.network.tor_controller.active_socks_port)
for p in ports:
if TorDetector.is_tor_port(p):
self.found_proxy.emit(("127.0.0.1", p))
break
else:
self.found_proxy.emit(
None
) # no proxy found, will hide the Tor checkbox
try:
stopq = self.stopQ.get(timeout=10.0) # keep trying every 10 seconds
if stopq is None:
return # we must have gotten a stop signal if we get here, break out of function, ending thread
# We were kicked, which means the tor port changed.
# Run the detection after a slight delay which increases the reliability.
QThread.msleep(250)
continue
except queue.Empty:
continue # timeout, keep looping
@staticmethod
def is_tor_port(port):
try:
s = (
socket._socketobject
if hasattr(socket, "_socketobject")
else socket.socket
)(socket.AF_INET, socket.SOCK_STREAM)
s.settimeout(0.1)
s.connect(("127.0.0.1", port))
# Tor responds uniquely to HTTP-like requests
s.send(b"GET\n")
if b"Tor is not an HTTP Proxy" in s.recv(1024):
return True
except socket.error:
pass
return False
diff --git a/electrum/electrumabc_gui/qt/utils/__init__.py b/electrum/electrumabc_gui/qt/utils/__init__.py
index 2caa6cea3..c863534ad 100644
--- a/electrum/electrumabc_gui/qt/utils/__init__.py
+++ b/electrum/electrumabc_gui/qt/utils/__init__.py
@@ -1,31 +1,30 @@
#!/usr/bin/env python3
#
# Electrum ABC - lightweight eCash client
# Copyright (C) 2020 The Electrum ABC Developers
# Copyright (C) 2019-2020 Axel Gembe
#
# Permission is hereby granted, free of charge, to any person
# obtaining a copy of this software and associated documentation files
# (the "Software"), to deal in the Software without restriction,
# including without limitation the rights to use, copy, modify, merge,
# publish, distribute, sublicense, and/or sell copies of the Software,
# and to permit persons to whom the Software is furnished to do so,
# subject to the following conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
# BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
# ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
# CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
from .aspect_layout import FixedAspectRatioLayout # noqa: F401
from .aspect_svg_widget import FixedAspectRatioSvgWidget # noqa: F401
from .color_utils import QColorLerp # noqa: F401
from .image_effect import ImageGraphicsEffect # noqa: F401
-from .user_port_validator import PortValidator, UserPortValidator # noqa: F401
diff --git a/electrum/electrumabc_gui/qt/utils/user_port_validator.py b/electrum/electrumabc_gui/qt/utils/user_port_validator.py
deleted file mode 100644
index b7b761be1..000000000
--- a/electrum/electrumabc_gui/qt/utils/user_port_validator.py
+++ /dev/null
@@ -1,76 +0,0 @@
-# Electrum ABC - lightweight eCash client
-# Copyright (C) 2020 The Electrum ABC Developers
-# Copyright (C) 2020 Axel Gembe
-#
-# Permission is hereby granted, free of charge, to any person
-# obtaining a copy of this software and associated documentation files
-# (the "Software"), to deal in the Software without restriction,
-# including without limitation the rights to use, copy, modify, merge,
-# publish, distribute, sublicense, and/or sell copies of the Software,
-# and to permit persons to whom the Software is furnished to do so,
-# subject to the following conditions:
-#
-# The above copyright notice and this permission notice shall be
-# included in all copies or substantial portions of the Software.
-#
-# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
-# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
-# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
-# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
-# BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
-# ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
-# CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
-# SOFTWARE.
-
-import warnings
-
-from PyQt5 import QtWidgets
-from PyQt5.QtCore import pyqtSignal
-from PyQt5.QtGui import QIntValidator, QValidator
-
-
-class PortValidator(QIntValidator):
- """A generic IP port validator. Accepts any number in the range [1,65535]
- by default."""
-
- stateChanged = pyqtSignal(QValidator.State)
-
- def __init__(self, parent, minimum=1, accept_zero=False):
- super().__init__(0, 65535, parent)
- if not isinstance(parent, QtWidgets.QLineEdit):
- warnings.warn(
- RuntimeWarning("PortValidator must be passed a QLineEdit parent")
- )
- self.minimum = minimum
- self.accept_zero = accept_zero
- self.stateChanged.connect(self.setRedBorder)
-
- def validate(self, inputStr: str, pos: int) -> QValidator.State:
- res = list(super().validate(inputStr, pos))
- if res[0] == QValidator.Acceptable:
- try:
- value = int(inputStr)
- if value < self.minimum and (not self.accept_zero or value != 0):
- res[0] = QValidator.Intermediate
- except (ValueError, TypeError):
- res[0] = QValidator.Invalid
- self.stateChanged.emit(res[0])
- return tuple(res)
-
- def setRedBorder(self, state):
- parent = self.parent()
- if isinstance(parent, QtWidgets.QLineEdit):
- if state == QValidator.Acceptable:
- parent.setStyleSheet("")
- else:
- parent.setStyleSheet("QLineEdit { border: 1px solid red }")
-
-
-class UserPortValidator(PortValidator):
- """
- Checks that a given port is either a high port (from 1024 to 65535) or zero.
- Additionally provides a callback for when the validation state changes.
- """
-
- def __init__(self, parent, accept_zero=False):
- super().__init__(parent, 1024, accept_zero)
diff --git a/electrum/electrumabc_gui/qt/utils/validators.py b/electrum/electrumabc_gui/qt/utils/validators.py
new file mode 100644
index 000000000..cc2490a09
--- /dev/null
+++ b/electrum/electrumabc_gui/qt/utils/validators.py
@@ -0,0 +1,156 @@
+# Electron Cash - lightweight Bitcoin client
+# Copyright (C) 2020, 2023 Axel Gembe
+#
+# Permission is hereby granted, free of charge, to any person
+# obtaining a copy of this software and associated documentation files
+# (the "Software"), to deal in the Software without restriction,
+# including without limitation the rights to use, copy, modify, merge,
+# publish, distribute, sublicense, and/or sell copies of the Software,
+# and to permit persons to whom the Software is furnished to do so,
+# subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be
+# included in all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
+# BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
+# ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+# CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+
+import ipaddress
+import re
+import warnings
+from typing import Tuple
+
+from PyQt5.QtCore import pyqtSignal
+from PyQt5.QtGui import QIntValidator, QValidator
+from PyQt5.QtWidgets import QLineEdit, QWidget
+
+
+def mark_widget(state: QValidator.State, widget: QWidget):
+ from ..util import ColorScheme
+
+ if state == QValidator.Acceptable:
+ widget.setStyleSheet("")
+ else:
+ widget.setStyleSheet(
+ "QWidget {{ border: 1px solid {color} }}".format(
+ color=ColorScheme.RED.get_html()
+ )
+ )
+
+
+# Based on: https://stackoverflow.com/a/33214423
+# By Eugene Yarmash (https://stackoverflow.com/users/244297)
+
+_all_numeric_re = re.compile(r"[0-9]+$")
+_allowed_label_re = re.compile(r"(?!-)[a-z0-9-]{1,63}(? 253:
+ return False
+
+ labels = hostname.split(".")
+
+ # the TLD must be not all-numeric
+ if _all_numeric_re.match(labels[-1]):
+ return False
+
+ return all(_allowed_label_re.match(label) for label in labels)
+
+
+class PortValidator(QIntValidator):
+ """A generic IP port validator. Accepts any number in the range [1,65535]
+ by default."""
+
+ stateChanged = pyqtSignal(QValidator.State)
+
+ def __init__(self, parent, minimum=1, accept_zero=False):
+ super().__init__(0, 65535, parent)
+ if not isinstance(parent, QLineEdit):
+ warnings.warn(
+ RuntimeWarning("PortValidator must be passed a QLineEdit parent")
+ )
+ self.minimum = minimum
+ self.accept_zero = accept_zero
+ self.stateChanged.connect(self.setRedBorder)
+
+ def validate(self, inputStr: str, pos: int) -> Tuple[QValidator.State, str, int]:
+ res = list(super().validate(inputStr, pos))
+ if res[0] == QValidator.Acceptable:
+ try:
+ value = int(inputStr)
+ if value < self.minimum and (not self.accept_zero or value != 0):
+ res[0] = QValidator.Intermediate
+ except (ValueError, TypeError):
+ res[0] = QValidator.Invalid
+ self.stateChanged.emit(res[0])
+ return tuple(res)
+
+ def setRedBorder(self, state):
+ parent = self.parent()
+ if isinstance(parent, QLineEdit):
+ mark_widget(state, parent)
+
+
+class UserPortValidator(PortValidator):
+ """
+ Checks that a given port is either a high port (from 1024 to 65535) or zero.
+ Additionally provides a callback for when the validation state changes.
+ """
+
+ def __init__(self, parent, accept_zero=False):
+ super().__init__(parent, 1024, accept_zero)
+
+
+class HostValidator(QValidator):
+ """
+ Validates a host string, accepts either IPV4, IPV6 or domain names
+ """
+
+ stateChanged = pyqtSignal(QValidator.State)
+
+ def __init__(self, parent):
+ super().__init__(parent)
+ if not isinstance(parent, QLineEdit):
+ warnings.warn(
+ RuntimeWarning("HostValidator must be passed a QLineEdit parent")
+ )
+ self.stateChanged.connect(self.setRedBorder)
+
+ def fixup(self, inputStr: str):
+ return inputStr.strip()
+
+ def validate(self, inputStr: str, pos: int) -> Tuple[QValidator.State, str, int]:
+ def set_state(state):
+ self.stateChanged.emit(state)
+ return state, inputStr, pos
+
+ if len(inputStr) == 0:
+ return set_state(QValidator.Intermediate)
+
+ # Check for valid IP address
+ try:
+ ipaddress.ip_address(inputStr)
+ return set_state(QValidator.Acceptable)
+ except ValueError:
+ pass
+
+ # Check for valid DNS
+ if is_valid_hostname(inputStr):
+ return set_state(QValidator.Acceptable)
+
+ return set_state(QValidator.Intermediate)
+
+ def setRedBorder(self, state):
+ parent = self.parent()
+ if isinstance(parent, QLineEdit):
+ mark_widget(state, parent)
diff --git a/electrum/electrumabc_plugins/fusion/qt.py b/electrum/electrumabc_plugins/fusion/qt.py
index f4df489c5..1f27145ff 100644
--- a/electrum/electrumabc_plugins/fusion/qt.py
+++ b/electrum/electrumabc_plugins/fusion/qt.py
@@ -1,2101 +1,2101 @@
#!/usr/bin/env python3
#
# Electron Cash - a lightweight Bitcoin Cash client
# CashFusion - an advanced coin anonymizer
#
# Copyright (C) 2020 Mark B. Lundeberg
#
# Permission is hereby granted, free of charge, to any person
# obtaining a copy of this software and associated documentation files
# (the "Software"), to deal in the Software without restriction,
# including without limitation the rights to use, copy, modify, merge,
# publish, distribute, sublicense, and/or sell copies of the Software,
# and to permit persons to whom the Software is furnished to do so,
# subject to the following conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
# BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
# ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
# CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
from __future__ import annotations
import threading
import weakref
from functools import partial
from typing import TYPE_CHECKING, Optional
from PyQt5 import QtWidgets
from PyQt5.QtCore import QMargins, QObject, QPoint, QSize, Qt, QTimer, pyqtSignal
from PyQt5.QtGui import QCursor, QIcon, QImage, QPainter
from electrumabc.address import Address
from electrumabc.i18n import _, ngettext
from electrumabc.plugins import hook, run_hook
from electrumabc.tor.controller import TorController
from electrumabc.util import InvalidPassword, do_in_main_thread, inv_dict
from electrumabc.wallet import AbstractWallet
from electrumabc_gui.qt.amountedit import XECAmountEdit
from electrumabc_gui.qt.main_window import ElectrumWindow
from electrumabc_gui.qt.popup_widget import KillPopupLabel, ShowPopupLabel
from electrumabc_gui.qt.statusbar import StatusBarButton
from electrumabc_gui.qt.tor_downloader import DownloadTorDialog
from electrumabc_gui.qt.util import (
Buttons,
CancelButton,
CloseButton,
ColorScheme,
OkButton,
PasswordLineEdit,
WaitingDialog,
WindowModalDialog,
)
-from electrumabc_gui.qt.utils import PortValidator, UserPortValidator
+from electrumabc_gui.qt.utils.validators import PortValidator, UserPortValidator
from .conf import Conf, Global
from .fusion import can_fuse_from, can_fuse_to
from .plugin import (
COIN_FRACTION_FUDGE_FACTOR,
MAX_LIMIT_FUSE_DEPTH,
TOR_PORTS,
FusionPlugin,
select_coins,
)
from .server import Params
from .util import get_coin_name
if TYPE_CHECKING:
from electrumabc_gui.qt.address_list import AddressList
from electrumabc_gui.qt import ElectrumGui
from pathlib import Path
heredir = Path(__file__).parent
# Lazy init the icons to avoid creating QObject when importing the module,
# in case the QApplication doesn't exist yet.
icon_fusion_logo = None
icon_fusion_logo_gray = None
image_red_exclamation = None
def get_icon_fusion_logo():
global icon_fusion_logo
if icon_fusion_logo is None:
icon_fusion_logo = QIcon(str(heredir / "Cash Fusion Logo - No Text.svg"))
return icon_fusion_logo
def get_icon_fusion_logo_gray():
global icon_fusion_logo_gray
if icon_fusion_logo_gray is None:
icon_fusion_logo_gray = QIcon(
str(heredir / "Cash Fusion Logo - No Text Gray.svg")
)
return icon_fusion_logo_gray
def get_image_red_exclamation():
global image_red_exclamation
if image_red_exclamation is None:
image_red_exclamation = QImage(str(heredir / "red_exclamation.png"))
return image_red_exclamation
class Plugin(FusionPlugin, QObject):
server_status_changed_signal = pyqtSignal(bool, tuple)
fusions_win = None
weak_settings_tab = None
gui: Optional[ElectrumGui] = None
initted = False
last_server_status = (True, ("Ok", ""))
_hide_history_txs = False
def __init__(self, *args, **kwargs):
# parentless top-level QObject. We need this type for the signal.
QObject.__init__(self)
# gives us self.config
FusionPlugin.__init__(self, *args, **kwargs)
# widgets we made, that need to be hidden & deleted when plugin is disabled
self.widgets = weakref.WeakSet()
self._hide_history_txs = Global(self.config).hide_history_txs
def on_close(self):
super().on_close()
# Shut down plugin.
# This can be triggered from one wallet's window while
# other wallets' windows have plugin-related modals open.
for window in self.gui.windows:
# this could be slow since it touches windows one by one... could optimize
# this by dispatching simultaneously.
self.on_close_window(window)
# Clean up
for w in self.widgets:
try:
w.setParent(None)
w.close()
w.hide()
w.deleteLater()
except Exception:
# could be but really we just want to suppress all exceptions
pass
# clean up member attributes to be tidy
# should trigger a deletion of object if not already dead
self.fusions_win = None
self.weak_settings_tab = None
self.gui = None
self.initted = False
@hook
def init_qt(self, gui):
# This gets called when this plugin is initialized, but also when
# any other plugin is initialized after us.
if self.initted:
return
self.initted = self.active = True # self.active is declared in super
self.gui = gui
if self.gui.nd:
# since a network dialog already exists, let's create the settings
# tab now.
self.on_network_dialog(self.gui.nd)
# We also have to find which windows are already open, and make
# them work with fusion.
for window in self.gui.windows:
self.on_new_window(window)
@hook
def address_list_context_menu_setup(self, address_list: AddressList, menu, addrs):
if not self.active:
return
wallet = address_list.wallet
window = address_list.main_window
network = wallet.network
if not (can_fuse_from(wallet) and can_fuse_to(wallet) and network):
return
if not hasattr(wallet, "_fusions"):
# that's a bug... all wallets should have this
return
coins = wallet.get_utxos(
addrs,
exclude_frozen=True,
mature=True,
confirmed_only=True,
exclude_slp=True,
)
def start_fusion():
def do_it(password):
try:
with wallet.lock:
if not hasattr(wallet, "_fusions"):
return
self.start_fusion(wallet, password, coins)
except RuntimeError as e:
window.show_error(
_("CashFusion failed: {error_message}").format(
error_message=str(e)
)
)
return
window.show_message(
ngettext(
"One coin has been sent to CashFusion for fusing.",
"{count} coins have been sent to CashFusion for fusing.",
len(coins),
).format(count=len(coins))
)
has_pw, password = Plugin.get_cached_pw(wallet)
if has_pw and password is None:
d = PasswordDialog(
wallet, _("Enter your password to fuse these coins"), do_it
)
d.show()
self.widgets.add(d)
else:
do_it(password)
if coins:
menu.addAction(
ngettext(
"Input one coin to CashFusion",
"Input {count} coins to CashFusion",
len(coins),
).format(count=len(coins)),
start_fusion,
)
@staticmethod
def get_spend_only_fused_coins_checkbox_attributes(wallet):
fuse_depth = Conf(wallet).fuse_depth
if fuse_depth > 0:
label = ngettext(
"Spend only fused coins, minimum {min} fusion",
"Spend only fused coins, minimum {min} fusions",
fuse_depth,
).format(min=fuse_depth)
tooltip = ngettext(
"If checked, only spend coins that have been anonymized by\n"
"CashFusion, after having been fused at least {min} time.",
"If checked, only spend coins that have been anonymized by\n"
"CashFusion, after having been fused at least {min} times.",
fuse_depth,
).format(min=fuse_depth)
else:
label = _("Spend only fused coins")
tooltip = _(
"If checked, only spend coins that have been\n"
"anonymized by CashFusion at least once."
)
return label, tooltip
@hook
def on_new_window(self, window):
# Called on initial plugin load (if enabled) and every new window; only once per window.
wallet = window.wallet
can_fuse = self.wallet_can_fuse(wallet)
if can_fuse:
sbbtn = FusionButton(self, wallet)
self.server_status_changed_signal.connect(sbbtn.update_server_error)
else:
# If we can not fuse we create a dummy fusion button that just displays a message
sbmsg = _(
"This wallet type ({wtype}) cannot be used with CashFusion.\n\n"
"Please use a standard deterministic spending wallet with CashFusion."
).format(wtype=wallet.wallet_type)
sbbtn = DisabledFusionButton(wallet, sbmsg)
# bit of a dirty hack, to insert our status bar icon (always using index 4, should put us just after the password-changer icon)
sb = window.statusBar()
sb.insertPermanentWidget(4, sbbtn)
self.widgets.add(sbbtn)
window._cashfusion_button = weakref.ref(sbbtn)
if not can_fuse:
# don't do anything with non-fusable wallets
# (if inter-wallet fusing is added, this should change.)
return
# NEW! Set up the send tab "Spend only fused coins" checkbox/control
if hasattr(window, "send_tab_extra_plugin_controls_hbox"):
hbox = window.send_tab_extra_plugin_controls_hbox
label, tooltip = self.get_spend_only_fused_coins_checkbox_attributes(wallet)
spend_only_fused_chk = QtWidgets.QCheckBox(label)
spend_only_fused_chk.setObjectName("spend_only_fused_chk")
spend_only_fused_chk.setToolTip(tooltip)
hbox.insertWidget(0, spend_only_fused_chk)
spend_only_fused_chk.setChecked(Conf(wallet).spend_only_fused_coins)
weak_window = weakref.ref(window)
def on_chk(b):
window = weak_window()
if window:
wallet = window.wallet
Conf(wallet).spend_only_fused_coins = b
window.do_update_fee() # trigger send tab to re-calculate things
spend_only_fused_chk.toggled.connect(on_chk)
self.widgets.add(spend_only_fused_chk)
want_autofuse = Conf(wallet).autofuse
self.add_wallet(wallet, window.gui_object.get_cached_password(wallet))
sbbtn.update_state()
# Set up the utxo_list column
self.patch_utxo_list(window.utxo_list)
# prompt for password if auto-fuse was enabled
if want_autofuse and not self.is_autofusing(wallet):
def callback(password):
self.enable_autofusing(wallet, password)
button = window._cashfusion_button()
if button:
button.update_state()
d = PasswordDialog(
wallet,
_(
"Previously you had auto-fusion enabled on this wallet. If you"
" would like to keep auto-fusing in the background, enter your"
" password."
),
callback_ok=callback,
)
d.show()
self.widgets.add(d)
@staticmethod
def patch_utxo_list(utxo_list):
if getattr(utxo_list, "_fusion_patched_", None) is not None:
return
header = utxo_list.headerItem()
header_labels = [header.text(i) for i in range(header.columnCount())]
header_labels.append(_("Fusion Status"))
utxo_list.update_headers(header_labels)
utxo_list._fusion_patched_ = header_labels[
-1
] # save the text to be able to find the column later
utxo_list.wallet.print_error("[fusion] Patched utxo_list")
@staticmethod
def find_utxo_list_fusion_column(utxo_list, *, just_column=False):
label_text = getattr(utxo_list, "_fusion_patched_", None)
if label_text is None:
return
header = utxo_list.headerItem()
if just_column:
# fast path, query just the column
col_ct = header.columnCount()
if not col_ct:
return None
# iterate in reverse (it's likely last)
for i in range(col_ct - 1, -1, -1):
if header.text(i) == label_text:
return i
else:
header_labels = [header.text(i) for i in range(header.columnCount())]
col = len(header_labels) - 1
# find the column, iterate in reverse since it's likely last
for i, lbl in enumerate(reversed(header_labels)):
if lbl == label_text:
col = len(header_labels) - 1 - i
break
return col, header, header_labels
@staticmethod
def unpatch_utxo_list(utxo_list):
tup = Plugin.find_utxo_list_fusion_column(utxo_list)
if tup is None:
return
col, header, header_labels = tup
del header_labels[col]
utxo_list.update_headers(header_labels)
delattr(utxo_list, "_fusion_patched_")
utxo_list.wallet.print_error("[fusion] Unpatched utxo_list")
@hook
def utxo_list_item_setup(self, utxo_list, item, utxo, name):
col = self.find_utxo_list_fusion_column(utxo_list, just_column=True)
if col is None:
return
wallet = utxo_list.wallet
fuse_depth = Conf(wallet).fuse_depth
is_fused = self.is_fuz_coin(wallet, utxo, require_depth=fuse_depth - 1)
is_partially_fused = (
is_fused if fuse_depth <= 1 else self.is_fuz_coin(wallet, utxo)
)
item.setIcon(col, QIcon())
if is_fused:
item.setText(col, _("Fused"))
item.setIcon(col, icon_fusion_logo)
elif is_partially_fused:
count = self.get_coin_fuz_count(wallet, utxo, require_depth=fuse_depth - 1)
item.setText(
col, _("Partial {count}/{total}").format(count=count, total=fuse_depth)
)
item.setIcon(col, icon_fusion_logo_gray)
elif self.is_fuz_address(wallet, utxo["address"], require_depth=fuse_depth - 1):
item.setText(col, _("Fuse Addr"))
item.setIcon(col, icon_fusion_logo)
item.setToolTip(
col,
_(
"This coin shares an address with a fused coin. Do not spend"
" separately."
),
)
elif utxo["height"] <= 0:
item.setText(col, _("Unconfirmed"))
elif utxo["coinbase"]:
# we disallow coinbase coins unconditionally -- due to miner feedback (they don't like shuffling these)
item.setText(col, _("Coinbase"))
else:
item.setText(col, _("Unfused"))
@hook
def spendable_coin_filter(self, window, coins):
"""Invoked by the send tab to filter out coins that aren't fused if the wallet has
'spend only fused coins' enabled."""
if not coins or not hasattr(window, "wallet"):
return
wallet = window.wallet
if not Conf(wallet).spend_only_fused_coins or not self.wallet_can_fuse(wallet):
return
# external_coins_addresses is only ever used if they are doing a sweep. in which case we always allow the coins
# involved in the sweep
external_coin_addresses = set()
if hasattr(window, "tx_external_keypairs"):
for pubkey in window.tx_external_keypairs:
a = Address.from_pubkey(pubkey)
external_coin_addresses.add(a)
# we can ONLY spend fused coins + ununfused living on a fused coin address
fuz_adrs_seen = set()
fuz_coins_seen = set()
with wallet.lock:
for coin in coins.copy():
if coin["address"] in external_coin_addresses:
# completely bypass this filter for external keypair dict
# which is only used for sweep dialog in send tab
continue
fuse_depth = Conf(wallet).fuse_depth
is_fuz_adr = self.is_fuz_address(
wallet, coin["address"], require_depth=fuse_depth - 1
)
if is_fuz_adr:
fuz_adrs_seen.add(coin["address"])
# we allow coins sitting on a fused address to be "spent as fused"
if (
not self.is_fuz_coin(wallet, coin, require_depth=fuse_depth - 1)
and not is_fuz_adr
):
coins.remove(coin)
else:
fuz_coins_seen.add(get_coin_name(coin))
# Force co-spending of other coins sitting on a fuzed address
for adr in fuz_adrs_seen:
adr_coins = wallet.get_addr_utxo(adr)
for name, adr_coin in adr_coins.items():
if (
name not in fuz_coins_seen
and not adr_coin["is_frozen_coin"]
and adr_coin.get("slp_token") is None
and not adr_coin.get("coinbase")
):
coins.append(adr_coin)
fuz_coins_seen.add(name)
@hook
def not_enough_funds_extra(self, window) -> Optional[str]:
"""Called by the Qt UI if there is a "not enough funds" error in the send tab"""
wallet = window.wallet
if not self.wallet_can_fuse(wallet):
return
conf = Conf(wallet)
if not conf.spend_only_fused_coins:
return
needs_fuz = [
coin
for coin in wallet.get_utxos(
exclude_frozen=True,
mature=True,
confirmed_only=bool(window.config.get("confirmed_only", False)),
)
if not self.is_fuz_coin(wallet, coin, require_depth=conf.fuse_depth - 1)
]
total = sum(c["value"] for c in needs_fuz)
n_coins = len(needs_fuz)
if total and needs_fuz:
return ngettext(
"{total_bch} in {n_coins} unfused coin",
"{total_bch} in {n_coins} unfused coins",
n_coins,
).format(
total_bch=window.format_amount(total) + " " + window.base_unit(),
n_coins=n_coins,
)
@hook
def on_close_window(self, window):
self.unpatch_utxo_list(window.utxo_list)
# Invoked when closing wallet or entire application
# Also called by on_close, above.
wallet = window.wallet
fusions = self.remove_wallet(wallet)
if not fusions:
return
for f in fusions:
f.stop("Closing wallet")
# Soft-stop background fuse if running.
# We avoid doing a hard disconnect in the middle of a fusion round.
# TODO: only do a gentler 'stop-if-waiting' and have a "STOP SOONER"
# button on the waiting dialog for the less patient users.
def task():
for f in fusions:
f.join()
d = WaitingDialog(
window.top_level_window(),
_("Shutting down active CashFusions (may take a minute to finish)"),
task,
)
d.exec_()
@hook
def on_new_password(self, window, old, new):
wallet = window.wallet
if self.is_autofusing(wallet):
try:
self.enable_autofusing(wallet, new)
self.print_error(wallet, "updated autofusion password")
except InvalidPassword:
self.disable_autofusing(wallet)
self.print_error(
wallet, "disabled autofusion due to incorrect password - BUG"
)
def show_util_window(
self,
):
if self.fusions_win is None:
# keep a singleton around
self.fusions_win = FusionsWindow(self)
self.widgets.add(self.fusions_win)
self.fusions_win.show()
self.fusions_win.raise_()
def requires_settings(self):
# called from main_window.py internal_plugins_dialog
return True
def settings_widget(self, window):
# called from main_window.py internal_plugins_dialog
btn = QtWidgets.QPushButton(_("Settings"))
btn.clicked.connect(self.show_settings_dialog)
return btn
def show_settings_dialog(self):
self.gui.show_network_dialog(None, jumpto="fusion")
@hook
def on_network_dialog(self, network_dialog):
if self.weak_settings_tab and self.weak_settings_tab():
return # already exists
settings_tab = SettingsWidget(self)
self.server_status_changed_signal.connect(settings_tab.update_server_error)
tabs = network_dialog.nlayout.tabs
tabs.addTab(settings_tab, get_icon_fusion_logo(), _("CashFusion"))
self.widgets.add(settings_tab)
self.weak_settings_tab = weakref.ref(settings_tab)
@hook
def on_network_dialog_jumpto(self, nlayout, location):
settings_tab = self.weak_settings_tab and self.weak_settings_tab()
if settings_tab and location in ("fusion", "cashfusion"):
nlayout.tabs.setCurrentWidget(settings_tab)
return True
def update_coins_ui(self, wallet):
"""Overrides super, the Fusion thread calls this in its thread context
to indicate it froze/unfroze some coins. We must update the coins tab,
but only in the main thread."""
def update_coins_tab(wallet):
strong_window = wallet and wallet.weak_window and wallet.weak_window()
if strong_window:
strong_window.utxo_list.update() # this is rate_limited so it's ok to call it many times in rapid succession.
do_in_main_thread(update_coins_tab, wallet)
def notify_server_status(self, b, tup):
"""Reimplemented from super"""
super().notify_server_status(b, tup)
status_tup = (b, tup)
if self.last_server_status != status_tup:
self.last_server_status = status_tup
self.server_status_changed_signal.emit(b, tup)
def get_server_error(self) -> tuple:
"""Returns a 2-tuple of strings for the last server error, or None
if there is no extant server error."""
if not self.last_server_status[0]:
return self.last_server_status[1]
@classmethod
def window_for_wallet(cls, wallet):
"""Convenience: Given a wallet instance, derefernces the weak_window
attribute of the wallet and returns a strong reference to the window.
May return None if the window is gone (deallocated)."""
assert isinstance(wallet, AbstractWallet)
return (wallet.weak_window and wallet.weak_window()) or None
@classmethod
def get_suitable_dialog_window_parent(cls, wallet_or_window):
"""Convenience: Given a wallet or a window instance, return a suitable
'top level window' parent to use for dialog boxes."""
if isinstance(wallet_or_window, AbstractWallet):
wallet = wallet_or_window
window = cls.window_for_wallet(wallet)
return (window and window.top_level_window()) or None
elif isinstance(wallet_or_window, ElectrumWindow):
window = wallet_or_window
return window.top_level_window()
else:
raise TypeError(
"Expected a wallet or a window instance, instead got"
f" {type(wallet_or_window)}"
)
@classmethod
def get_cached_pw(cls, wallet):
"""Will return a tuple: (bool, password) for the given wallet. The
boolean is whether the wallet is password protected and the second
item is the cached password, if it's known, otherwise None if it is not
known. If the wallet has no password protection the tuple is always
(False, None)."""
if not wallet.has_password():
return False, None
window = cls.window_for_wallet(wallet)
if not window:
raise RuntimeError(
f"Wallet {wallet.diagnostic_name()} lacks a valid ElectrumWindow"
" instance!"
)
pw = window.gui_object.get_cached_password(wallet)
if pw is not None:
try:
wallet.check_password(pw)
except InvalidPassword:
pw = None
return True, pw
@classmethod
def cache_pw(cls, wallet, password):
window = cls.window_for_wallet(wallet)
if window:
window.gui_object.cache_password(wallet, password)
def enable_autofusing(self, wallet, password):
"""Overrides super, if super successfully turns on autofusing, kicks
off the timer to check that Tor is working."""
super().enable_autofusing(wallet, password)
if self.is_autofusing(wallet):
# ok, autofuse enable success -- kick of the timer task to check if
# Tor is good
do_in_main_thread(
self._maybe_prompt_user_if_they_want_integrated_tor_if_no_tor_found,
wallet,
)
_integrated_tor_timer = None
def _maybe_download_tor(
self, window: ElectrumWindow, tor_controller: TorController
) -> bool:
"""Ask the user if he wants to download Tor, then download it.
If the download was successful or Tor was installed with a package manager,
return True.
"""
icon_pm = get_icon_fusion_logo().pixmap(32)
answer = window.question(
_(
"CashFusion requires Tor to operate anonymously. Would you like to "
"download the Tor client now? Alternatively (preferably), you can "
"install Tor for Linux and MacOS using your system's package manager "
"(in this case click the No button when installing is complete)."
),
icon=icon_pm,
title=_("Tor Required"),
parent=None,
app_modal=True,
rich_text=True,
defaultButton=QtWidgets.QMessageBox.Yes,
)
if answer:
dialog = DownloadTorDialog(self.config, window)
dialog.exec_()
# Check Tor availability no matter if the download succeeded, as the user
# could have installed Tor with his package manager while the question dialog
# was running.
if tor_controller.detect_tor():
return True
return False
def _maybe_prompt_user_if_they_want_integrated_tor_if_no_tor_found(
self, wallet: AbstractWallet
):
if self._integrated_tor_timer:
# timer already active or already prompted user
return
weak_self = weakref.ref(self)
weak_window = wallet.weak_window
if not weak_window or not weak_window():
# Something's wrong -- no window for wallet
return
network = self.gui.daemon.network
tbt = network.tor_controller.tor_binary_type
if tbt == TorController.BinaryType.MISSING:
if not self._maybe_download_tor(weak_window(), network.tor_controller):
# Tor unavailable and download failed or cancelled
return
def chk_tor_ok():
self: Optional[Plugin] = weak_self()
if not self:
return
self._integrated_tor_timer = None # kill QTimer reference
window = weak_window()
if (
not window
or not self.active
or not self.gui
or not self.gui.windows
or self.tor_port_good is not None
):
return
if (
not network
or not network.tor_controller.is_available()
or network.tor_controller.is_enabled()
):
return
icon_pm = get_icon_fusion_logo().pixmap(32)
answer = window.question(
_(
"CashFusion requires Tor to operate anonymously. Would"
" you like to enable the Tor client now?"
),
icon=icon_pm,
title=_("Tor Required"),
parent=None,
app_modal=True,
rich_text=True,
defaultButton=QtWidgets.QMessageBox.Yes,
)
if not answer:
return
def on_status(controller):
try:
network.tor_controller.status_changed.remove(
on_status
) # remove the callback immediately
except ValueError:
pass
if controller.status == controller.Status.STARTED:
buttons = [_("Settings..."), _("Ok")]
index = window.show_message(
_("The Tor client has been successfully started."),
detail_text=(
_(
"The Tor client can be stopped at any time from the"
" Network Settings -> Proxy Tab, however CashFusion"
" does require Tor in order to operate correctly."
)
),
icon=icon_pm,
rich_text=True,
buttons=buttons,
defaultButton=buttons[1],
escapeButton=buttons[1],
)
if index == 0:
# They want to go to "Settings..." so send
# them to the Tor settings (in Proxy tab)
self.gui.show_network_dialog(window, jumpto="tor")
else:
controller.set_enabled(
False
) # latch it back to False so we may prompt them again in the future
window.show_error(_("There was an error starting the Tor client"))
network.tor_controller.status_changed.append(on_status)
network.tor_controller.set_enabled(True)
self._integrated_tor_timer = t = QTimer()
# if in 5 seconds no tor port, ask user if they want to enable the Tor
t.timeout.connect(chk_tor_ok)
t.setSingleShot(True)
t.start(2500)
@hook
def history_list_filter(self, history_list, h_item, label):
# NB: 'h_item' might be None due to performance reasons
if self._hide_history_txs:
return bool(
label.startswith("CashFusion ")
) # this string is not translated for performance reasons
return None
@hook
def history_list_context_menu_setup(self, history_list, menu, item, tx_hash):
# NB: We unconditionally create this menu if the plugin is loaded because
# it's possible for any wallet, even a watching-only wallet to have
# fusion tx's with the correct labels (if the user has imported labels).
menu.addSeparator()
def action_callback():
self._hide_history_txs = not self._hide_history_txs
Global(self.config).hide_history_txs = self._hide_history_txs
action.setChecked(self._hide_history_txs)
if self._hide_history_txs:
tip = _("Fusion transactions are now hidden")
else:
tip = _("Fusion transactions are now shown")
QtWidgets.QToolTip.showText(QCursor.pos(), tip, history_list)
# unconditionally update this history list as it may be embedded in the
# address_detail window and not a global history list..
history_list.update()
for w in self.gui.windows:
# Need to update all the other open windows.
# Note: We still miss any other open windows' address-detail
# history lists with this.. but that's ok as most of the
# time it won't be noticed by people and actually
# finding all those windows would just make this code
# less maintainable.
# check if not already updated above
if history_list is not w.history_list:
w.history_list.update()
action = menu.addAction(_("Hide CashFusions"), action_callback)
action.setCheckable(True)
action.setChecked(self._hide_history_txs)
class PasswordDialog(WindowModalDialog):
"""Slightly fancier password dialog -- can be used non-modal (asynchronous) and has internal password checking.
To run non-modally, use .show with the callbacks; to run modally, use .run."""
def __init__(self, wallet, message, callback_ok=None, callback_cancel=None):
parent = Plugin.get_suitable_dialog_window_parent(wallet)
super().__init__(parent=parent, title=_("Enter Password"))
self.setWindowIcon(get_icon_fusion_logo())
self.wallet = wallet
self.callback_ok = callback_ok
self.callback_cancel = callback_cancel
self.password = None
vbox = QtWidgets.QVBoxLayout(self)
self.msglabel = QtWidgets.QLabel(message)
self.msglabel.setWordWrap(True)
self.msglabel.setMinimumWidth(250)
self.msglabel.setSizePolicy(
QtWidgets.QSizePolicy.MinimumExpanding, QtWidgets.QSizePolicy.Expanding
)
hbox = QtWidgets.QHBoxLayout()
iconlabel = QtWidgets.QLabel()
iconlabel.setPixmap(get_icon_fusion_logo().pixmap(32))
hbox.addWidget(iconlabel)
hbox.addWidget(self.msglabel, 1, Qt.AlignLeft | Qt.AlignVCenter)
cmargins = hbox.contentsMargins()
cmargins.setBottom(10)
hbox.setContentsMargins(cmargins) # pad the bottom a bit
vbox.addLayout(hbox, 1)
self.pwle = PasswordLineEdit()
grid_for_hook_api = QtWidgets.QGridLayout()
grid_for_hook_api.setContentsMargins(0, 0, 0, 0)
grid_for_hook_api.addWidget(self.pwle, 0, 0)
run_hook(
"password_dialog", self.pwle, grid_for_hook_api, 0
) # this is for the virtual keyboard plugin
vbox.addLayout(grid_for_hook_api)
self.badpass_msg = (
"" + _("Incorrect password entered. Please try again.") + ""
)
buttons = QtWidgets.QHBoxLayout()
buttons.addStretch(1)
buttons.addWidget(CancelButton(self))
okbutton = OkButton(self)
okbutton.clicked.disconnect()
okbutton.clicked.connect(self.pw_entered)
buttons.addWidget(okbutton)
vbox.addLayout(buttons)
def _on_pw_ok(self, password):
self.password = password
Plugin.cache_pw(
self.wallet, password
) # to remember it for a time so as to not keep bugging the user
self.accept()
if self.callback_ok:
self.callback_ok(password)
def _chk_pass(self, password):
pw_ok = not self.wallet.has_password()
if not pw_ok:
try:
self.wallet.check_password(password)
pw_ok = True
except InvalidPassword:
pass
return pw_ok
def pw_entered(
self,
):
password = self.pwle.text()
if self._chk_pass(password):
self._on_pw_ok(password)
else:
self.msglabel.setText(self.badpass_msg)
self.pwle.clear()
self.pwle.setFocus()
def closeEvent(self, event):
"""This happens if .run() is called, then dialog is closed."""
super().closeEvent(event)
if event.isAccepted():
self._close_hide_common()
def hideEvent(self, event):
"""This happens if .show() is called, then dialog is closed."""
super().hideEvent(event)
if event.isAccepted():
self._close_hide_common()
def _close_hide_common(self):
if not self.result() and self.callback_cancel:
self.callback_cancel(self)
self.setParent(None)
self.deleteLater()
def run(self):
self.exec_()
return self.password
class DisabledFusionButton(StatusBarButton):
def __init__(self, wallet, message):
super().__init__(get_icon_fusion_logo_gray(), "Fusion")
self.wallet = wallet
self.message = message
self.setToolTip(_("CashFusion (disabled)"))
self.clicked.connect(self.show_message)
def show_message(self):
QtWidgets.QMessageBox.information(
Plugin.get_suitable_dialog_window_parent(self.wallet),
_("CashFusion is disabled"),
self.message,
)
class FusionButton(StatusBarButton):
def __init__(self, plugin, wallet):
super().__init__(QIcon(), "Fusion")
self.clicked.connect(self.toggle_autofuse)
self.plugin = plugin
self.wallet = wallet
self.server_error: tuple = None
self.icon_autofusing_on = get_icon_fusion_logo()
self.icon_autofusing_off = get_icon_fusion_logo_gray()
self.icon_fusing_problem = self.style().standardIcon(
QtWidgets.QStyle.SP_MessageBoxWarning
)
# title = QWidgetAction(self)
# title.setDefaultWidget(QtWidgets.QLabel("" + _("CashFusion") + ""))
self.action_toggle = QtWidgets.QAction(_("Auto-Fuse in Background"))
self.action_toggle.setCheckable(True)
self.action_toggle.triggered.connect(self.toggle_autofuse)
action_separator1 = QtWidgets.QAction(self)
action_separator1.setSeparator(True)
action_wsettings = QtWidgets.QAction(_("Wallet Fusion Settings..."), self)
action_wsettings.triggered.connect(self.show_wallet_settings)
action_settings = QtWidgets.QAction(_("Server Settings..."), self)
action_settings.triggered.connect(self.plugin.show_settings_dialog)
action_separator2 = QtWidgets.QAction(self)
action_separator2.setSeparator(True)
action_util = QtWidgets.QAction(_("Fusions..."), self)
action_util.triggered.connect(self.plugin.show_util_window)
self.addActions(
[
self.action_toggle,
action_separator1,
action_wsettings,
action_settings,
action_separator2,
action_util,
]
)
self.setContextMenuPolicy(Qt.ActionsContextMenu)
self.update_state()
def update_state(self):
autofuse = self.plugin.is_autofusing(self.wallet)
self.action_toggle.setChecked(autofuse)
if autofuse:
self.setIcon(self.icon_autofusing_on)
self.setToolTip(_("CashFusion is fusing in the background for this wallet"))
self.setStatusTip(_("CashFusion Auto-fusion - Enabled"))
else:
self.setIcon(self.icon_autofusing_off)
self.setToolTip(
_("Auto-fusion is paused for this wallet (click to enable)")
)
self.setStatusTip(_("CashFusion Auto-fusion - Disabled (click to enable)"))
if self.server_error:
self.setToolTip(_("CashFusion") + ": " + ", ".join(self.server_error))
self.setStatusTip(_("CashFusion") + ": " + ", ".join(self.server_error))
def paintEvent(self, event):
super().paintEvent(event)
if event.isAccepted() and self.server_error:
# draw error overlay if we are in an error state
p = QPainter(self)
try:
p.setClipRegion(event.region())
r = self.rect()
r -= QMargins(4, 6, 4, 6)
r.moveCenter(r.center() + QPoint(4, 4))
p.drawImage(r, get_image_red_exclamation())
finally:
# paranoia. The above never raises but.. if it does.. PyQt will
# crash hard if we don't end the QPainter properly before
# returning.
p.end()
del p
def toggle_autofuse(self):
plugin = self.plugin
autofuse = plugin.is_autofusing(self.wallet)
if not autofuse:
has_pw, password = Plugin.get_cached_pw(self.wallet)
if has_pw and password is None:
# Fixme: See if we can not use a blocking password dialog here.
pd = PasswordDialog(
self.wallet,
_(
"To perform auto-fusion in the background, please enter your"
" password."
),
)
# just in case this plugin is unloaded while this dialog is up
self.plugin.widgets.add(pd)
password = pd.run()
del pd
# must check plugin.active because user can theoretically kill plugin
# from another window while the above password dialog is up
if password is None or not plugin.active:
return
try:
plugin.enable_autofusing(self.wallet, password)
except InvalidPassword:
"""Somehow the password changed from underneath us. Silenty ignore."""
else:
running = plugin.disable_autofusing(self.wallet)
if running:
res = QtWidgets.QMessageBox.question(
Plugin.get_suitable_dialog_window_parent(self.wallet),
_("Disabling automatic Cash Fusions"),
_(
"New automatic fusions will not be started, but you have {num}"
" currently in progress. Would you like to signal them to stop?"
).format(num=len(running)),
)
if res == QtWidgets.QMessageBox.Yes:
for f in running:
f.stop("Stop requested by user")
self.update_state()
def show_wallet_settings(self):
win = getattr(self.wallet, "_cashfusion_settings_window", None)
if not win:
win = WalletSettingsDialog(
Plugin.get_suitable_dialog_window_parent(self.wallet),
self.plugin,
self.wallet,
)
# ensures if plugin is unloaded while dialog is up, that the dialog will
# be killed.
self.plugin.widgets.add(win)
win.show()
win.raise_()
def update_server_error(self):
tup = self.plugin.get_server_error()
changed = tup != self.server_error
if not changed:
return
self.server_error = tup
# make sure name is unique per FusionButton widget
name = "CashFusionError;" + str(id(self))
if self.server_error:
weak_plugin = weakref.ref(self.plugin)
def onClick():
KillPopupLabel(name)
plugin = weak_plugin()
if plugin:
plugin.show_settings_dialog()
ShowPopupLabel(
name=name,
text="{}
{}".format(
_("Server Error"), _("Click this popup to resolve")
),
target=self,
timeout=20000,
onClick=onClick,
onRightClick=onClick,
dark_mode=ColorScheme.dark_scheme,
)
else:
KillPopupLabel(name)
self.update() # causes a repaint
window = self.wallet.weak_window and self.wallet.weak_window()
if window:
window.print_error(
"CashFusion server_error is now {}".format(self.server_error)
)
oldTip = self.statusTip()
self.update_state()
newTip = self.statusTip()
if newTip != oldTip:
window.statusBar().showMessage(newTip, 7500)
class SettingsWidget(QtWidgets.QWidget):
torscanthread = None
torscanthread_update = pyqtSignal(object)
def __init__(self, plugin, parent=None):
super().__init__(parent)
self.plugin = plugin
self.torscanthread_ping = threading.Event()
self.torscanthread_update.connect(self.torport_update)
main_layout = QtWidgets.QVBoxLayout(self)
box = QtWidgets.QGroupBox(_("Network"))
main_layout.addWidget(box, 0, Qt.AlignTop | Qt.AlignHCenter)
slayout = QtWidgets.QVBoxLayout(box)
grid = QtWidgets.QGridLayout()
slayout.addLayout(grid)
grid.addWidget(QtWidgets.QLabel(_("Server")), 0, 0)
hbox = QtWidgets.QHBoxLayout()
grid.addLayout(hbox, 0, 1)
self.combo_server_host = QtWidgets.QComboBox()
self.combo_server_host.setEditable(True)
self.combo_server_host.setInsertPolicy(QtWidgets.QComboBox.NoInsert)
self.combo_server_host.setCompleter(None)
self.combo_server_host.setSizePolicy(
QtWidgets.QSizePolicy.Expanding, QtWidgets.QSizePolicy.Fixed
)
self.combo_server_host.activated.connect(self.combo_server_activated)
self.combo_server_host.lineEdit().textEdited.connect(self.user_changed_server)
self.combo_server_host.addItems(
[
f'{s[0]} ({s[1]}{" - ssl" if s[2] else ""})'
for s in Global.Defaults.ServerList
]
)
hbox.addWidget(self.combo_server_host)
hbox.addWidget(QtWidgets.QLabel(_("P:")))
self.le_server_port = QtWidgets.QLineEdit()
self.le_server_port.setMaximumWidth(50)
self.le_server_port.setValidator(PortValidator(self.le_server_port))
self.le_server_port.textEdited.connect(self.user_changed_server)
hbox.addWidget(self.le_server_port)
self.cb_server_ssl = QtWidgets.QCheckBox(_("SSL"))
self.cb_server_ssl.clicked.connect(self.user_changed_server)
hbox.addWidget(self.cb_server_ssl)
self.server_error_label = QtWidgets.QLabel()
self.server_error_label.setAlignment(Qt.AlignTop | Qt.AlignJustify)
grid.addWidget(self.server_error_label, 1, 0, 1, -1)
grid.addWidget(QtWidgets.QLabel(_("Tor")), 2, 0)
hbox = QtWidgets.QHBoxLayout()
grid.addLayout(hbox, 2, 1)
self.le_tor_host = QtWidgets.QLineEdit("localhost")
self.le_tor_host.textEdited.connect(self.user_edit_torhost)
hbox.addWidget(self.le_tor_host)
hbox.addWidget(QtWidgets.QLabel(_("P:")))
self.le_tor_port = QtWidgets.QLineEdit()
self.le_tor_port.setMaximumWidth(50)
self.le_tor_port.setValidator(UserPortValidator(self.le_tor_port))
self.le_tor_port.textEdited.connect(self.user_edit_torport)
hbox.addWidget(self.le_tor_port)
self.l_tor_status = QtWidgets.QLabel()
hbox.addWidget(self.l_tor_status)
self.b_tor_refresh = QtWidgets.QPushButton()
self.b_tor_refresh.clicked.connect(self.torscanthread_ping.set)
self.b_tor_refresh.setIcon(
self.style().standardIcon(QtWidgets.QStyle.SP_BrowserReload)
)
self.b_tor_refresh.setDefault(False)
self.b_tor_refresh.setAutoDefault(False)
hbox.addWidget(self.b_tor_refresh)
self.cb_tor_auto = QtWidgets.QCheckBox(_("Autodetect"))
self.cb_tor_auto.clicked.connect(self.cb_tor_auto_clicked)
hbox.addWidget(self.cb_tor_auto)
btn = QtWidgets.QPushButton(_("Fusions..."))
btn.setDefault(False)
btn.setAutoDefault(False)
btn.clicked.connect(self.plugin.show_util_window)
buts = Buttons(btn)
buts.setAlignment(Qt.AlignRight | Qt.AlignTop)
main_layout.addLayout(buts)
main_layout.addStretch(1)
self.stretch_item_index = main_layout.count() - 1
self.server_widget = ServerWidget(self.plugin)
self.server_widget.layout().setContentsMargins(0, 0, 0, 0)
main_layout.addWidget(self.server_widget)
self.timer_server_widget_visibility = QTimer(self.server_widget)
self.timer_server_widget_visibility.setSingleShot(False)
self.timer_server_widget_visibility.timeout.connect(
self.update_server_widget_visibility
)
self.server_widget_index = main_layout.count() - 1
self.pm_good_proxy = QIcon(":icons/status_connected_proxy.svg").pixmap(24)
self.pm_bad_proxy = QIcon(":icons/status_disconnected.svg").pixmap(24)
def update_server(self):
# called initially / when config changes
host, port, ssl = self.plugin.get_server()
try: # see if it's in default list, if so we can set it ...
index = Global.Defaults.ServerList.index((host, port, ssl))
except ValueError: # not in list
index = -1
self.combo_server_host.setCurrentIndex(index)
self.combo_server_host.setEditText(host)
self.le_server_port.setText(str(port))
self.cb_server_ssl.setChecked(ssl)
def update_server_error(self):
errtup = self.plugin.get_server_error()
self.server_error_label.setHidden(errtup is None)
if errtup:
color = ColorScheme.RED.get_html()
self.server_error_label.setText(
f'{errtup[0]}: {errtup[1]}'
)
def combo_server_activated(self, index):
# only triggered when user selects a combo item
self.plugin.set_server(*Global.Defaults.ServerList[index])
self.update_server()
def user_changed_server(self, *args):
# user edited the host / port / ssl
host = self.combo_server_host.currentText()
try:
port = int(self.le_server_port.text())
except ValueError:
port = 0
ssl = self.cb_server_ssl.isChecked()
self.plugin.set_server(host, port, ssl)
def update_tor(
self,
):
# called on init an switch of auto
autoport = self.plugin.has_auto_torport()
host = self.plugin.get_torhost()
port = self.plugin.get_torport()
self.l_tor_status.clear()
self.torport_update(port)
self.cb_tor_auto.setChecked(autoport)
self.le_tor_host.setEnabled(not autoport)
self.le_tor_host.setText(str(host))
self.le_tor_port.setEnabled(not autoport)
if not autoport:
self.le_tor_port.setText(str(port))
def torport_update(self, goodport):
# signalled from the tor checker thread
autoport = self.plugin.has_auto_torport()
port = self.plugin.get_torport()
if autoport:
sport = "?" if port is None else str(port)
self.le_tor_port.setText(sport)
if goodport is None:
self.l_tor_status.setPixmap(self.pm_bad_proxy)
# TODO: switch from %-string to str.format (update also translation files)
if autoport:
self.l_tor_status.setToolTip(
_("Cannot find a Tor proxy on ports %(ports)s.")
% {"ports": TOR_PORTS}
)
else:
self.l_tor_status.setToolTip(
_("Cannot find a Tor proxy on port %(port)d.") % {"port": port}
)
else:
self.l_tor_status.setToolTip(_("Found a valid Tor proxy on this port."))
self.l_tor_status.setPixmap(self.pm_good_proxy)
def user_edit_torhost(self, host):
self.plugin.set_torhost(host)
self.torscanthread_ping.set()
def user_edit_torport(self, sport):
try:
port = int(sport)
except ValueError:
return
self.plugin.set_torport(port)
self.torscanthread_ping.set()
def cb_tor_auto_clicked(self, state):
self.plugin.set_torport("auto" if state else "manual")
port = self.plugin.get_torport()
if port is not None:
self.le_tor_port.setText(str(port))
self.torscanthread_ping.set()
self.update_tor()
def refresh(self):
self.update_server()
self.update_tor()
self.update_server_widget_visibility()
self.update_server_error()
def update_server_widget_visibility(self):
if not self.server_widget.is_server_running():
self.server_widget.setHidden(True)
self.layout().setStretch(self.stretch_item_index, 1)
self.layout().setStretch(self.server_widget_index, 0)
else:
self.server_widget.setHidden(False)
self.layout().setStretch(self.stretch_item_index, 0)
self.layout().setStretch(self.server_widget_index, 1)
def showEvent(self, event):
super().showEvent(event)
if not event.isAccepted():
return
self.refresh()
self.timer_server_widget_visibility.start(2000)
if self.torscanthread is None:
self.torscanthread = threading.Thread(
name="Fusion-scan_torport_settings", target=self.scan_torport_loop
)
self.torscanthread.daemon = True
self.torscanthread_stopping = False
self.torscanthread.start()
def _hide_close_common(self):
self.timer_server_widget_visibility.stop()
self.torscanthread_stopping = True
self.torscanthread_ping.set()
self.torscanthread = None
def closeEvent(self, event):
super().closeEvent(event)
if not event.isAccepted():
return
self._hide_close_common()
def hideEvent(self, event):
super().hideEvent(event)
if not event.isAccepted():
return
self._hide_close_common()
def scan_torport_loop(
self,
):
while not self.torscanthread_stopping:
goodport = self.plugin.scan_torport()
self.torscanthread_update.emit(goodport)
self.torscanthread_ping.wait(10)
self.torscanthread_ping.clear()
class WalletSettingsDialog(WindowModalDialog):
GUI_DEFAULT_FUSE_DEPTH = (
3 # This what the fuse depth spinbox defaults to, if checked (on new installs)
)
def __init__(self, parent, plugin, wallet):
super().__init__(parent=parent, title=_("CashFusion - Wallet Settings"))
self.setWindowIcon(get_icon_fusion_logo())
self.plugin = plugin
self.wallet = wallet
self.conf = Conf(self.wallet)
self.idx2confkey = {} # int -> 'normal', 'consolidate', etc..
self.confkey2idx = {} # str 'normal', 'consolidate', etc -> int
assert not hasattr(self.wallet, "_cashfusion_settings_window")
main_window = self.wallet.weak_window()
assert main_window
self.wallet._cashfusion_settings_window = self
main_layout = QtWidgets.QVBoxLayout(self)
hbox = QtWidgets.QHBoxLayout()
hbox.addWidget(QtWidgets.QLabel(_("Fusion mode:")))
self.mode_cb = mode_cb = QtWidgets.QComboBox()
hbox.addWidget(mode_cb)
main_layout.addLayout(hbox)
self.gb_fuse_depth = gb = QtWidgets.QGroupBox(_("Fusion Rounds"))
gb.setToolTip(
_(
"If checked, CashFusion will fuse each coin this many times.\n"
"If unchecked, Cashfusion will fuse indefinitely until paused."
)
)
hbox = QtWidgets.QHBoxLayout(gb)
self.chk_fuse_depth = chk = QtWidgets.QCheckBox(_("Fuse coins this many times"))
hbox.addWidget(chk, 1)
self.sb_fuse_depth = sb = QtWidgets.QSpinBox()
sb.setRange(1, MAX_LIMIT_FUSE_DEPTH)
if self.conf.fuse_depth <= 0:
# Default it to this if unchecked
self.sb_fuse_depth.setValue(self.GUI_DEFAULT_FUSE_DEPTH)
sb.setMinimumWidth(75)
hbox.addWidget(sb)
chk.toggled.connect(self.edited_fuse_depth)
sb.valueChanged.connect(self.edited_fuse_depth)
main_layout.addWidget(gb)
self.gb_coinbase = gb = QtWidgets.QGroupBox(_("Coinbase Coins"))
vbox = QtWidgets.QVBoxLayout(gb)
self.cb_coinbase = QtWidgets.QCheckBox(
_("Auto-fuse coinbase coins (if mature)")
)
self.cb_coinbase.clicked.connect(self._on_cb_coinbase)
vbox.addWidget(self.cb_coinbase)
# The coinbase-related group box is hidden by default. It becomes
# visible permanently when the wallet settings dialog has seen at least
# one coinbase coin, indicating a miner's wallet. For most users the
# coinbase checkbox is confusing, which is why we prefer to hide it.
gb.setHidden(True)
main_layout.addWidget(gb)
box = QtWidgets.QGroupBox(_("Self-Fusing"))
main_layout.addWidget(box)
slayout = QtWidgets.QVBoxLayout(box)
lbl = QtWidgets.QLabel(
_("Allow this wallet to participate multiply in the same fusion round?")
)
lbl.setWordWrap(True)
slayout.addWidget(lbl)
box = QtWidgets.QHBoxLayout()
box.setContentsMargins(0, 0, 0, 0)
self.combo_self_fuse = QtWidgets.QComboBox()
self.combo_self_fuse.addItem(_("No"), 1)
self.combo_self_fuse.addItem(_("Yes - as up to two players"), 2)
box.addStretch(1)
box.addWidget(self.combo_self_fuse)
slayout.addLayout(box)
del box
self.combo_self_fuse.activated.connect(self.chose_self_fuse)
self.stacked_layout = stacked_layout = QtWidgets.QStackedLayout()
main_layout.addLayout(stacked_layout)
# Stacked Layout pages ...
# Normal
normal_page_w = QtWidgets.QWidget()
normal_page_layout = QtWidgets.QVBoxLayout(normal_page_w)
self.confkey2idx["normal"] = stacked_layout.addWidget(normal_page_w)
mode_cb.addItem(_("Normal"))
lbl = QtWidgets.QLabel("- " + _("Normal mode") + " -")
lbl.setAlignment(Qt.AlignCenter)
normal_page_layout.addWidget(lbl)
# Consolidate
consolidate_page_w = QtWidgets.QWidget()
consolidate_page_layout = QtWidgets.QVBoxLayout(consolidate_page_w)
self.confkey2idx["consolidate"] = stacked_layout.addWidget(consolidate_page_w)
mode_cb.addItem(_("Consolidate"))
lbl = QtWidgets.QLabel("- " + _("Consolidation mode") + " -")
lbl.setAlignment(Qt.AlignCenter)
consolidate_page_layout.addWidget(lbl)
# Fan-out
fanout_page_w = QtWidgets.QWidget()
fanout_page_layout = QtWidgets.QVBoxLayout(fanout_page_w)
self.confkey2idx["fan-out"] = stacked_layout.addWidget(fanout_page_w)
mode_cb.addItem(_("Fan-out"))
lbl = QtWidgets.QLabel("- " + _("Fan-out mode") + " -")
lbl.setAlignment(Qt.AlignCenter)
fanout_page_layout.addWidget(lbl)
# Custom
self.custom_page_w = custom_page_w = QtWidgets.QWidget()
custom_page_layout = QtWidgets.QVBoxLayout(custom_page_w)
custom_page_layout.setContentsMargins(0, 0, 0, 0)
self.confkey2idx["custom"] = stacked_layout.addWidget(custom_page_w)
mode_cb.addItem(_("Custom"))
mode_cb.currentIndexChanged.connect(
self._on_mode_changed
) # intentionally connected after all items already added
box = QtWidgets.QGroupBox(_("Auto-Fusion Coin Selection"))
custom_page_layout.addWidget(box)
slayout = QtWidgets.QVBoxLayout(box)
grid = QtWidgets.QGridLayout()
slayout.addLayout(grid)
self.radio_select_size = QtWidgets.QRadioButton(
_("Target typical output amount")
)
grid.addWidget(self.radio_select_size, 0, 0)
self.radio_select_fraction = QtWidgets.QRadioButton(_("Per-coin random chance"))
grid.addWidget(self.radio_select_fraction, 1, 0)
self.radio_select_count = QtWidgets.QRadioButton(
_("Target number of coins in wallet")
)
grid.addWidget(self.radio_select_count, 2, 0)
self.radio_select_size.clicked.connect(self.edited_size)
self.radio_select_fraction.clicked.connect(self.edited_fraction)
self.radio_select_count.clicked.connect(self.edited_count)
self.amt_selector_size = XECAmountEdit(main_window.get_decimal_point())
grid.addWidget(self.amt_selector_size, 0, 1)
self.sb_selector_fraction = QtWidgets.QDoubleSpinBox()
self.sb_selector_fraction.setRange(0.1, 100.0)
self.sb_selector_fraction.setSuffix("%")
self.sb_selector_fraction.setDecimals(1)
grid.addWidget(self.sb_selector_fraction, 1, 1)
self.sb_selector_count = QtWidgets.QSpinBox()
self.sb_selector_count.setRange(
COIN_FRACTION_FUDGE_FACTOR, 9999
) # Somewhat hardcoded limit of 9999 is arbitrary, have this come from constants?
grid.addWidget(self.sb_selector_count, 2, 1)
self.amt_selector_size.editingFinished.connect(self.edited_size)
self.sb_selector_fraction.valueChanged.connect(self.edited_fraction)
self.sb_selector_count.valueChanged.connect(self.edited_count)
# Clicking the radio button should bring its corresponding widget buddy into focus
self.radio_select_size.clicked.connect(self.amt_selector_size.setFocus)
self.radio_select_fraction.clicked.connect(self.sb_selector_fraction.setFocus)
self.radio_select_count.clicked.connect(self.sb_selector_count.setFocus)
low_warn_blurb = _("Are you trying to consolidate?")
low_warn_tooltip = _("Click for consolidation tips")
low_warn_blurb_link = '' + low_warn_blurb + ""
self.l_warn_selection = QtWidgets.QLabel(
"" + low_warn_blurb_link + ""
)
self.l_warn_selection.setToolTip(low_warn_tooltip)
self.l_warn_selection.linkActivated.connect(self._show_low_warn_help)
self.l_warn_selection.setAlignment(Qt.AlignJustify | Qt.AlignVCenter)
qs = QtWidgets.QSizePolicy(
QtWidgets.QSizePolicy.Preferred, QtWidgets.QSizePolicy.Preferred
)
qs.setRetainSizeWhenHidden(True)
self.l_warn_selection.setSizePolicy(qs)
slayout.addWidget(self.l_warn_selection)
slayout.setAlignment(self.l_warn_selection, Qt.AlignCenter)
box = QtWidgets.QGroupBox(_("Auto-Fusion Limits"))
custom_page_layout.addWidget(box)
slayout = QtWidgets.QVBoxLayout(box)
grid = QtWidgets.QGridLayout()
slayout.addLayout(grid)
grid.addWidget(QtWidgets.QLabel(_("Number of queued fusions")), 0, 0)
self.sb_queued_autofuse = QtWidgets.QSpinBox()
self.sb_queued_autofuse.setRange(
1, 10
) # hard-coded rande 1-10, maybe have this come from some constants?
self.sb_queued_autofuse.setMinimumWidth(
50
) # just so it doesn't end up too tiny
grid.addWidget(self.sb_queued_autofuse, 0, 1)
self.cb_autofuse_only_all_confirmed = QtWidgets.QCheckBox(
_("Only auto-fuse when all coins are confirmed")
)
slayout.addWidget(self.cb_autofuse_only_all_confirmed)
grid.addWidget(QtWidgets.QWidget(), 0, 2)
grid.setColumnStretch(2, 1) # spacer
self.sb_queued_autofuse.valueChanged.connect(self.edited_queued_autofuse)
self.cb_autofuse_only_all_confirmed.clicked.connect(self.clicked_confirmed_only)
# / end pages
cbut = CloseButton(self)
main_layout.addLayout(Buttons(cbut))
cbut.setDefault(False)
cbut.setAutoDefault(False)
self.idx2confkey = inv_dict(
self.confkey2idx
) # This must be set-up before this function returns
# We do this here in addition to in showEvent because on some platforms
# (such as macOS), the window animates-in before refreshing properly and
# then it refreshes, leading to a jumpy glitch. If we do this, it
# slides-in already looking as it should.
self.refresh()
def _show_low_warn_help(self):
low_warn_message = (
_("If you wish to consolidate coins:")
+ ""
+ "- "
+ _("Specify a maximum of 1 queued fusion")
+ "
- "
+ _("Set 'self-fusing' to 'No'")
+ "
- "
+ _("Check the 'only when all coins are confirmed' checkbox")
+ "
"
+ _(
"If you do not wish to necessarily consolidate coins, then it's"
" perfectly acceptable to ignore this tip."
)
)
self.show_message(low_warn_message, title=_("Help"), rich_text=True)
def _on_mode_changed(self, idx: int):
self.conf.fusion_mode = self.idx2confkey[
idx
] # will raise on bad idx, which indicates programming error.
self.refresh()
def _on_cb_coinbase(self, checked: bool):
self.conf.autofuse_coinbase = checked
self.refresh()
def _maybe_switch_page(self):
mode = self.conf.fusion_mode
try:
idx = self.confkey2idx[mode]
idx_custom = self.confkey2idx["custom"]
# The below conditional ensures that the custom page always
# disappears from the layout if not selected. We do this because it
# is rather large and makes this window unnecessarily big. Note this
# only works if the 'custom' page is last.. otherwise bad things
# happen!
assert idx_custom == max(
self.confkey2idx.values()
) # ensures custom is last page otherwise this code breaks
if idx == idx_custom:
if not self.stacked_layout.itemAt(idx_custom):
self.stacked_layout.insertWidget(idx_custom, self.custom_page_w)
elif self.stacked_layout.count() - 1 == idx_custom:
self.stacked_layout.takeAt(idx_custom)
self.stacked_layout.setCurrentIndex(idx)
self.mode_cb.setCurrentIndex(idx)
except KeyError as e:
# should never happen because settings object filters out unknown modes
raise RuntimeError(f"INTERNAL ERROR: Unknown fusion mode: '{mode}'") from e
self.updateGeometry()
self.resize(self.sizeHint())
return idx == idx_custom
def refresh(self):
eligible, ineligible, sum_value, has_unconfirmed, has_coinbase = select_coins(
self.wallet
)
select_type, select_amount = self.conf.selector
edit_widgets = [
self.amt_selector_size,
self.sb_selector_fraction,
self.sb_selector_count,
self.sb_queued_autofuse,
self.cb_autofuse_only_all_confirmed,
self.combo_self_fuse,
self.stacked_layout,
self.mode_cb,
self.cb_coinbase,
self.sb_fuse_depth,
self.chk_fuse_depth,
]
try:
for w in edit_widgets:
# Block spurious editingFinished signals and valueChanged signals as
# we modify the state and focus of widgets programatically below.
# On macOS not doing this led to a very strange/spazzy UI.
w.blockSignals(True)
self.cb_coinbase.setChecked(self.conf.autofuse_coinbase)
if not self.gb_coinbase.isVisible():
cb_latch = self.conf.coinbase_seen_latch
if cb_latch or self.cb_coinbase.isChecked() or has_coinbase:
if not cb_latch:
# Once latched to true, this UI element will forever be
# visible for this wallet. It means the wallet is a miner's
# wallet and they care about coinbase coins.
self.conf.coinbase_seen_latch = True
self.gb_coinbase.setHidden(False)
del cb_latch
is_custom_page = self._maybe_switch_page()
idx = 0
if self.conf.self_fuse_players > 1:
idx = 1
self.combo_self_fuse.setCurrentIndex(idx)
del idx
if self.conf.fuse_depth > 0:
self.sb_fuse_depth.setValue(self.conf.fuse_depth)
self.chk_fuse_depth.setChecked(self.conf.fuse_depth > 0)
self.sb_fuse_depth.setEnabled(self.conf.fuse_depth > 0)
if is_custom_page:
self.amt_selector_size.setEnabled(select_type == "size")
self.sb_selector_count.setEnabled(select_type == "count")
self.sb_selector_fraction.setEnabled(select_type == "fraction")
if select_type == "size":
self.radio_select_size.setChecked(True)
sel_size = select_amount
if sum_value > 0:
sel_fraction = min(
COIN_FRACTION_FUDGE_FACTOR * select_amount / sum_value, 1.0
)
else:
sel_fraction = 1.0
elif select_type == "count":
self.radio_select_count.setChecked(True)
sel_size = max(sum_value / max(select_amount, 1), 10000)
sel_fraction = COIN_FRACTION_FUDGE_FACTOR / max(select_amount, 1)
elif select_type == "fraction":
self.radio_select_fraction.setChecked(True)
sel_size = max(
sum_value * select_amount / COIN_FRACTION_FUDGE_FACTOR, 10000
)
sel_fraction = select_amount
else:
self.conf.selector = None
return self.refresh()
sel_count = round(COIN_FRACTION_FUDGE_FACTOR / max(sel_fraction, 0.001))
self.amt_selector_size.setAmount(round(sel_size))
self.sb_selector_fraction.setValue(
max(min(sel_fraction, 1.0), 0.001) * 100.0
)
self.sb_selector_count.setValue(sel_count)
try:
self.sb_queued_autofuse.setValue(self.conf.queued_autofuse)
except (TypeError, ValueError):
pass # should never happen but paranoia pays off in the long-term
conf_only = self.conf.autofuse_confirmed_only
self.cb_autofuse_only_all_confirmed.setChecked(conf_only)
self.l_warn_selection.setVisible(
sel_fraction > 0.2
and (not conf_only or self.sb_queued_autofuse.value() > 1)
)
finally:
# re-enable signals
for w in edit_widgets:
w.blockSignals(False)
def edited_size(
self,
):
size = self.amt_selector_size.get_amount()
if size is None or size < 10000:
size = 10000
self.conf.selector = ("size", size)
self.refresh()
def edited_fraction(
self,
):
fraction = max(self.sb_selector_fraction.value() / 100.0, 0.0)
self.conf.selector = ("fraction", round(fraction, 3))
self.refresh()
def edited_count(
self,
):
count = self.sb_selector_count.value()
self.conf.selector = ("count", count)
self.refresh()
def edited_queued_autofuse(
self,
):
prevval = self.conf.queued_autofuse
numfuse = self.sb_queued_autofuse.value()
self.conf.queued_autofuse = numfuse
if prevval > numfuse:
for f in list(self.wallet._fusions_auto):
f.stop("User decreased queued-fuse limit", not_if_running=True)
self.refresh()
def edited_fuse_depth(
self,
):
prevval = self.conf.fuse_depth
newval = self.sb_fuse_depth.value() if self.chk_fuse_depth.isChecked() else 0
self.conf.fuse_depth = newval
if prevval == 0 or (prevval > newval and newval != 0):
for f in list(self.wallet._fusions_auto):
f.stop("User decreased fuse depth limit", not_if_running=False)
# update the send tab label for the "spend only confirmed coins" checkbox
main_window = self.wallet.weak_window and self.wallet.weak_window()
if main_window:
chk = main_window.findChild(
QtWidgets.QCheckBox, "spend_only_fused_chk", Qt.FindChildrenRecursively
)
if chk:
(
label,
tooltip,
) = self.plugin.get_spend_only_fused_coins_checkbox_attributes(
self.wallet
)
chk.setText(label)
chk.setToolTip(tooltip)
# Coins tab may need redisplay if we changed these settings
if prevval != newval:
main_window.utxo_list.update()
self.refresh()
def clicked_confirmed_only(self, checked):
self.conf.autofuse_confirmed_only = checked
self.refresh()
def chose_self_fuse(
self,
):
sel = self.combo_self_fuse.currentData()
oldsel = self.conf.self_fuse_players
if oldsel != sel:
self.conf.self_fuse_players = sel
for f in self.wallet._fusions:
# we have to stop waiting fusions since the tags won't overlap.
# otherwise, the user will end up self fusing way too much.
f.stop("User changed self-fuse limit", not_if_running=True)
self.refresh()
def closeEvent(self, event):
super().closeEvent(event)
if event.isAccepted():
self.setParent(None)
del self.wallet._cashfusion_settings_window
def showEvent(self, event):
super().showEvent(event)
if event.isAccepted():
self.refresh()
class ServerFusionsBaseMixin:
def __init__(self, plugin, refresh_interval=2000):
assert isinstance(self, QtWidgets.QWidget)
self.plugin = plugin
self.refresh_interval = refresh_interval
self.timer_refresh = QTimer(self)
self.timer_refresh.setSingleShot(False)
self.timer_refresh.timeout.connect(self.refresh)
def _on_show(self):
self.timer_refresh.start(self.refresh_interval)
self.refresh()
def _on_hide(self):
self.timer_refresh.stop()
def showEvent(self, event):
super().showEvent(event)
if event.isAccepted():
self._on_show()
def hideEvent(self, event):
super().hideEvent(event)
if event.isAccepted():
self._on_hide()
def closeEvent(self, event):
super().closeEvent(event)
if event.isAccepted():
self._on_hide()
def refresh(self):
raise NotImplementedError(
"ServerFusionsBaseMixin refresh() needs an implementation"
)
class ServerWidget(ServerFusionsBaseMixin, QtWidgets.QWidget):
def __init__(self, plugin, parent=None):
QtWidgets.QWidget.__init__(self, parent)
ServerFusionsBaseMixin.__init__(self, plugin)
main_layout = QtWidgets.QVBoxLayout(self)
self.serverbox = QtWidgets.QGroupBox(_("Server"))
main_layout.addWidget(self.serverbox)
# self.serverbox.setSizePolicy(QSizePolicy.Preferred, QSizePolicy.Preferred)
slayout = QtWidgets.QVBoxLayout(self.serverbox)
self.l_server_status = QtWidgets.QLabel()
slayout.addWidget(self.l_server_status)
self.t_server_waiting = QtWidgets.QTableWidget()
self.t_server_waiting.setColumnCount(3)
self.t_server_waiting.setRowCount(len(Params.tiers))
self.t_server_waiting.setHorizontalHeaderLabels(
[_("Tier (sats)"), _("Num players"), ""]
)
for i, t in enumerate(Params.tiers):
button = QtWidgets.QPushButton(_("Start"))
button.setDefault(False)
# on some platforms if we don't do this, one of the buttons traps "Enter"
# key
button.setAutoDefault(False)
button.clicked.connect(partial(self.clicked_start_fuse, t))
self.t_server_waiting.setCellWidget(i, 2, button)
slayout.addWidget(self.t_server_waiting)
def sizeHint(self):
return QSize(300, 150)
def refresh(self):
if self.is_server_running():
self.t_server_waiting.setEnabled(True)
self.l_server_status.setText(
_("Server status: ACTIVE")
+ f" {self.plugin.fusion_server.host}:{self.plugin.fusion_server.port}"
)
table = self.t_server_waiting
table.setRowCount(len(self.plugin.fusion_server.waiting_pools))
for i, (t, pool) in enumerate(
self.plugin.fusion_server.waiting_pools.items()
):
table.setItem(i, 0, QtWidgets.QTableWidgetItem(str(t)))
table.setItem(i, 1, QtWidgets.QTableWidgetItem(str(len(pool.pool))))
else:
self.t_server_waiting.setEnabled(False)
self.l_server_status.setText(_("Server status: NOT RUNNING"))
def is_server_running(self):
return bool(self.plugin.fusion_server)
def clicked_start_fuse(self, tier, event):
if self.plugin.fusion_server is None:
return
self.plugin.fusion_server.start_fuse(tier)
class FusionsWindow(ServerFusionsBaseMixin, QtWidgets.QDialog):
def __init__(self, plugin):
QtWidgets.QDialog.__init__(self, parent=None)
ServerFusionsBaseMixin.__init__(self, plugin, refresh_interval=1000)
self.setWindowTitle(_("CashFusion - Fusions"))
self.setWindowIcon(get_icon_fusion_logo())
main_layout = QtWidgets.QVBoxLayout(self)
clientbox = QtWidgets.QGroupBox(_("Fusions"))
main_layout.addWidget(clientbox)
clayout = QtWidgets.QVBoxLayout(clientbox)
self.t_active_fusions = QtWidgets.QTreeWidget()
self.t_active_fusions.setHeaderLabels(
[_("Wallet"), _("Status"), _("Status Extra")]
)
self.t_active_fusions.setContextMenuPolicy(Qt.CustomContextMenu)
self.t_active_fusions.customContextMenuRequested.connect(
self.create_menu_active_fusions
)
self.t_active_fusions.setSelectionMode(
QtWidgets.QAbstractItemView.ExtendedSelection
)
self.t_active_fusions.itemDoubleClicked.connect(self.on_double_clicked)
clayout.addWidget(self.t_active_fusions)
self.resize(520, 240) # TODO: Have this somehow not be hard-coded
def refresh(self):
tree = self.t_active_fusions
reselect_fusions = {i.data(0, Qt.UserRole)() for i in tree.selectedItems()}
reselect_fusions.discard(None)
reselect_items = []
tree.clear()
for fusion in reversed(self.plugin.get_all_fusions()):
wname = fusion.target_wallet.diagnostic_name()
status, status_ext = fusion.status
item = QtWidgets.QTreeWidgetItem([wname, status, status_ext])
item.setToolTip(0, wname) # this doesn't always fit in the column
item.setToolTip(2, status_ext or "") # neither does this
item.setData(0, Qt.UserRole, weakref.ref(fusion))
if fusion in reselect_fusions:
reselect_items.append(item)
tree.addTopLevelItem(item)
for item in reselect_items:
item.setSelected(True)
def create_menu_active_fusions(self, position):
selected = self.t_active_fusions.selectedItems()
if not selected:
return
fusions = {i.data(0, Qt.UserRole)() for i in selected}
fusions.discard(None)
statuses = {f.status[0] for f in fusions}
selection_of_1_fusion = list(fusions)[0] if len(fusions) == 1 else None
has_live = "running" in statuses or "waiting" in statuses
menu = QtWidgets.QMenu()
def cancel():
for fusion in fusions:
fusion.stop(_("Stop requested by user"))
if has_live:
if "running" in statuses:
msg = _("Cancel (at end of round)")
else:
msg = _("Cancel")
menu.addAction(msg, cancel)
if selection_of_1_fusion and selection_of_1_fusion.txid:
menu.addAction(
_("View Tx..."), lambda: self._open_tx_for_fusion(selection_of_1_fusion)
)
if not menu.isEmpty():
menu.exec_(self.t_active_fusions.viewport().mapToGlobal(position))
def on_double_clicked(self, item, column):
self._open_tx_for_fusion(item.data(0, Qt.UserRole)())
def _open_tx_for_fusion(self, fusion):
if not fusion or not fusion.txid or not fusion.target_wallet:
return
wallet = fusion.target_wallet
window = wallet.weak_window and wallet.weak_window()
txid = fusion.txid
if window:
tx = window.wallet.transactions.get(txid)
if tx:
window.show_transaction(tx, wallet.get_label(txid))
else:
window.show_error(_("Transaction not yet in wallet"))