purple-electrumwallet/electrum/gui/qml/qedaemon.py

import base64
import os
import threading
from typing import TYPE_CHECKING

from PyQt6.QtCore import Qt, QAbstractListModel, QModelIndex
from PyQt6.QtCore import pyqtProperty, pyqtSignal, pyqtSlot, QObject

from electrum.i18n import _
from electrum.logging import get_logger
from electrum.util import WalletFileException, standardize_path, InvalidPassword, send_exception_to_crash_reporter
from electrum.plugin import run_hook
from electrum.lnchannel import ChannelState
from electrum.bitcoin import is_address
from electrum.bitcoin import verify_usermessage_with_address
from electrum.storage import StorageReadWriteError

from .auth import AuthMixin, auth_protect
from .qefx import QEFX
from .qewallet import QEWallet
from .qewizard import QENewWalletWizard, QEServerConnectWizard, QETermsOfUseWizard

if TYPE_CHECKING:
    from electrum.daemon import Daemon
    from electrum.plugin import Plugins


# wallet list model. supports both wallet basenames (wallet file basenames)
# and whole Wallet instances (loaded wallets)
from .util import check_password_strength


class QEWalletListModel(QAbstractListModel):
    _logger = get_logger(__name__)

    # define listmodel rolemap
    _ROLE_NAMES= ('name', 'path', 'active')
    _ROLE_KEYS = range(Qt.ItemDataRole.UserRole, Qt.ItemDataRole.UserRole + len(_ROLE_NAMES))
    _ROLE_MAP  = dict(zip(_ROLE_KEYS, [bytearray(x.encode()) for x in _ROLE_NAMES]))

    def __init__(self, daemon: 'Daemon', parent=None):
        QAbstractListModel.__init__(self, parent)
        self.daemon = daemon
        self._wallets = []
        self.reload()

    def rowCount(self, index):
        return len(self._wallets)

    def roleNames(self):
        return self._ROLE_MAP

    def data(self, index, role):
        (wallet_name, wallet_path) = self._wallets[index.row()]
        role_index = role - Qt.ItemDataRole.UserRole
        role_name = self._ROLE_NAMES[role_index]
        if role_name == 'name':
            return wallet_name
        if role_name == 'path':
            return wallet_path
        if role_name == 'active':
            return self.daemon.get_wallet(wallet_path) is not None

    @pyqtSlot()
    def reload(self):
        self._logger.debug('enumerating available wallets')
        self.beginResetModel()
        self._wallets = []
        self.endResetModel()

        available = []
        wallet_folder = os.path.dirname(self.daemon.config.get_wallet_path())
        with os.scandir(wallet_folder) as it:
            for i in it:
                if i.is_file() and not i.name.startswith('.'):
                    available.append(i.path)
        for path in sorted(available):
            wallet = self.daemon.get_wallet(path)
            self.add_wallet(wallet_path=path)

    def add_wallet(self, wallet_path):
        self.beginInsertRows(QModelIndex(), len(self._wallets), len(self._wallets))
        wallet_name = os.path.basename(wallet_path)
        wallet_path = standardize_path(wallet_path)
        item = (wallet_name, wallet_path)
        self._wallets.append(item)
        self.endInsertRows()

    def remove_wallet(self, path):
        i = 0
        wallets = []
        remove = -1
        for wallet_name, wallet_path in self._wallets:
            if wallet_path == path:
                remove = i
            else:
                wallets.append((wallet_name, wallet_path))
            i += 1

        if remove >= 0:
            self.beginRemoveRows(QModelIndex(), remove, remove)
            self._wallets = wallets
            self.endRemoveRows()

    @pyqtSlot(str, result=bool)
    def wallet_name_exists(self, name):
        for wallet_name, wallet_path in self._wallets:
            if name == wallet_name:
                return True
        return False

    @pyqtSlot(str)
    def updateWallet(self, path):
        i = 0
        for wallet_name, wallet_path in self._wallets:
            if wallet_path == path:
                mi = self.createIndex(i, i)
                self.dataChanged.emit(mi, mi, self._ROLE_KEYS)
                return
            i += 1


class QEDaemon(AuthMixin, QObject):
    instance = None  # type: Optional[QEDaemon]

    _logger = get_logger(__name__)

    _available_wallets = None
    _current_wallet = None
    _new_wallet_wizard = None
    _terms_of_use_wizard = None
    _server_connect_wizard = None
    _path = None
    _name = None
    _use_single_password = False
    _password = None
    _loading = False

    _backendWalletLoaded = pyqtSignal([str], arguments=['password'])

    availableWalletsChanged = pyqtSignal()
    fxChanged = pyqtSignal()
    newWalletWizardChanged = pyqtSignal()
    termsOfUseWizardChanged = pyqtSignal()
    serverConnectWizardChanged = pyqtSignal()
    loadingChanged = pyqtSignal()
    requestNewPassword = pyqtSignal()

    walletLoaded = pyqtSignal([str, str], arguments=['name', 'path'])
    walletRequiresPassword = pyqtSignal([str, str], arguments=['name', 'path'])
    walletOpenError = pyqtSignal([str], arguments=["error"])
    walletDeleteError = pyqtSignal([str, str], arguments=['code', 'message'])

    def __init__(self, daemon: 'Daemon', plugins: 'Plugins', parent=None):
        super().__init__(parent)
        if QEDaemon.instance:
            raise RuntimeError('There should only be one QEDaemon instance')
        QEDaemon.instance = self
        self.daemon = daemon
        self.plugins = plugins
        self.qefx = QEFX(daemon.fx, daemon.config)

        self._backendWalletLoaded.connect(self._on_backend_wallet_loaded)

    @pyqtSlot()
    def passwordValidityCheck(self):
        if not self._walletdb._validPassword:
            self.walletRequiresPassword.emit(self._name, self._path)

    @pyqtSlot()
    @pyqtSlot(str)
    @pyqtSlot(str, str)
    def loadWallet(self, path=None, password=None):
        if self._loading:
            return
        self._loading = True

        if path is None:
            self._path = self.daemon.config.get('wallet_path')  # command line -w option
            if self._path is None:
                self._path = self.daemon.config.CURRENT_WALLET
        else:
            self._path = path
        if self._path is None:
            self._loading = False
            return

        self.loadingChanged.emit()

        self._path = standardize_path(self._path)
        self._name = os.path.basename(self._path)

        self._logger.debug('load wallet ' + str(self._path))

        # password unification helper:
        # - if pw not given (None), try pw of current wallet.
        # - but "" empty str passwords are kept as-is, to open passwordless wallets
        if password is None:
            password = self._password

        # map explicit empty str password to None. the backend disallows empty str passwords.
        if password == '':
            password = None

        wallet_already_open = self.daemon.get_wallet(self._path)
        if wallet_already_open is not None:
            password = QEWallet.getInstanceFor(wallet_already_open).password

        def load_wallet_task():
            success = False
            try:
                local_password = password  # need this in local scope
                wallet = None
                try:
                    wallet = self.daemon.load_wallet(
                        self._path,
                        password=local_password,
                        upgrade=True,
                        # might have a keystore password, but unencrypted storage. we want to prompt for pw even then:
                        force_check_password=True,
                    )
                except InvalidPassword:
                    self.walletRequiresPassword.emit(self._name, self._path)
                except FileNotFoundError:
                    self.walletOpenError.emit(_('File not found') + f":\n{self._path}")
                except StorageReadWriteError:
                    self.walletOpenError.emit(_('Could not read/write file'))
                except WalletFileException as e:
                    self.walletOpenError.emit(_('Could not open wallet: {}').format(str(e)))
                    if e.should_report_crash:
                        send_exception_to_crash_reporter(e)

                if wallet is None:
                    return

                if self.daemon.config.WALLET_SHOULD_USE_SINGLE_PASSWORD:
                    self._use_single_password = self._update_password_for_directory_and_unlock_wallets(old_password=local_password, new_password=local_password)
                    if not self._use_single_password and self.daemon.config.WALLET_ANDROID_USE_BIOMETRIC_AUTHENTICATION:
                        # we need to disable biometric auth if the user creates wallets with different passwords as
                        # we only store one encrypted password which is not associated to a specific wallet
                        self._logger.warning(f"disabling biometric authentication, not in single password mode")
                        self.daemon.config.WALLET_ANDROID_USE_BIOMETRIC_AUTHENTICATION = False
                        self.daemon.config.WALLET_ANDROID_BIOMETRIC_AUTH_WRAPPED_WALLET_PASSWORD = ''
                        self.daemon.config.WALLET_ANDROID_BIOMETRIC_AUTH_ENCRYPTED_WRAP_KEY = ''
                    self._password = local_password
                    self.singlePasswordChanged.emit()
                    self._logger.info(f'use single password: {self._use_single_password}')
                else:
                    self._logger.info('use single password disabled by config')
                self.daemon.config.WALLET_DID_USE_SINGLE_PASSWORD = self._use_single_password

                run_hook('load_wallet', wallet)

                success = True
                self._backendWalletLoaded.emit(local_password)
            finally:
                if not success:  # if successful, _loading guard will be reset by _on_backend_wallet_loaded
                    self._loading = False
                    self.loadingChanged.emit()

        threading.Thread(target=load_wallet_task, daemon=False).start()

    @pyqtSlot()
    @pyqtSlot(str)
    def _on_backend_wallet_loaded(self, password=None):
        self._logger.debug('_on_backend_wallet_loaded')
        wallet = self.daemon.get_wallet(self._path)
        assert wallet is not None
        self._current_wallet = QEWallet.getInstanceFor(wallet)
        self.availableWallets.updateWallet(self._path)
        wallet.unlock(password or None)  # not conditional on wallet.requires_unlock in qml, as
        # the auth wrapper doesn't pass the entered password, but instead we rely on the password in memory
        self._loading = False
        self.loadingChanged.emit()
        self.walletLoaded.emit(self._name, self._path)

    @pyqtSlot(QEWallet)
    @pyqtSlot(QEWallet, bool)
    @pyqtSlot(QEWallet, bool, bool)
    def checkThenDeleteWallet(self, wallet, confirm_requests=False, confirm_balance=False):
        if wallet.wallet.lnworker:
            lnchannels = wallet.wallet.lnworker.get_channel_objects()
            if any([channel.get_state() != ChannelState.REDEEMED and not channel.is_backup() for channel in lnchannels.values()]):
                self.walletDeleteError.emit('unclosed_channels', _('There are still channels that are not fully closed'))
                return

        num_requests = len(wallet.wallet.get_unpaid_requests())
        if num_requests > 0 and not confirm_requests:
            self.walletDeleteError.emit('unpaid_requests', _('There are still unpaid requests. Really delete?'))
            return

        c, u, x = wallet.wallet.get_balance()
        if c+u+x > 0 and not wallet.wallet.is_watching_only() and not confirm_balance:
            self.walletDeleteError.emit('balance', _('There are still coins present in this wallet. Really delete?'))
            return

        self.delete_wallet(wallet)

    @auth_protect(message=_('Really delete this wallet?'))
    def delete_wallet(self, wallet):
        path = standardize_path(wallet.wallet.storage.path)
        self._logger.debug('deleting wallet with path %s' % path)
        self._current_wallet = None
        # TODO walletLoaded signal is confusing
        self.walletLoaded.emit(None, None)

        if not self.daemon.delete_wallet(path):
            self.walletDeleteError.emit('error', _('Problem deleting wallet'))
            return

        self.availableWallets.remove_wallet(path)

    @pyqtProperty(bool, notify=loadingChanged)
    def loading(self):
        return self._loading

    @pyqtProperty(QEWallet, notify=walletLoaded)
    def currentWallet(self):
        return self._current_wallet

    @pyqtProperty(QEWalletListModel, notify=availableWalletsChanged)
    def availableWallets(self):
        if not self._available_wallets:
            self._available_wallets = QEWalletListModel(self.daemon)

        return self._available_wallets

    @pyqtProperty(QEFX, notify=fxChanged)
    def fx(self):
        return self.qefx

    @pyqtSlot(str, result=list)
    def getWalletsUnlockableWithPassword(self, password: str) -> list[str]:
        """
        Returns any wallet that can be unlocked with the given password.
        Can be used as fallback to unlock another wallet the user entered a
        password that doesn't work for the current wallet but might work for another one.
        """
        wallet_dir = os.path.dirname(self.daemon.config.get_wallet_path())
        _, _, wallet_paths_can_unlock = self.daemon.check_password_for_directory(
            old_password=password,
            new_password=None,
            wallet_dir=wallet_dir,
        )
        if not wallet_paths_can_unlock:
            return []
        self._logger.debug(f"getWalletsUnlockableWithPassword: can unlock {len(wallet_paths_can_unlock)} wallets")
        return [str(path) for path in wallet_paths_can_unlock]

    @pyqtSlot(str, result=int)
    def numWalletsWithPassword(self, password: str) -> int:
        """Returns the number of wallets that can be unlocked with the given password"""
        wallet_paths_can_unlock = self.getWalletsUnlockableWithPassword(password)
        return len(wallet_paths_can_unlock)

    singlePasswordChanged = pyqtSignal()
    @pyqtProperty(bool, notify=singlePasswordChanged)
    def singlePasswordEnabled(self):
        """
        singlePasswordEnabled is False if:
            a.) the user has no wallet (and password) yet
            b.) the user has wallets with different passwords (legacy)
            c.) all wallets are locked, we couldn't check yet if they all use the same password
            d.) we are on desktop where different passwords are allowed
        """
        return self._use_single_password

    @pyqtProperty(str, notify=singlePasswordChanged)
    def singlePassword(self):
        """
        self._password is also set to the last loaded wallet password if we WANT a single password,
        but don't actually have a single password yet. So singlePassword being set doesn't strictly
        mean all wallets use the same password.
        """
        return self._password

    @singlePassword.setter
    def singlePassword(self, password: str):
        assert password
        assert self.daemon.config.WALLET_SHOULD_USE_SINGLE_PASSWORD
        if self._password != password:
            self._password = password
            self.singlePasswordChanged.emit()

    @pyqtSlot(result=str)
    def suggestWalletName(self):
        # FIXME why not use util.get_new_wallet_name ?
        i = 1
        while self.availableWallets.wallet_name_exists(f'wallet_{i}'):
            i = i + 1
        return f'wallet_{i}'

    @pyqtSlot()
    @auth_protect(method='wallet_password_only')
    def startChangePassword(self):
        if self._use_single_password:
            self.requestNewPassword.emit()
        else:
            self.currentWallet.requestNewPassword.emit()

    @pyqtSlot(str, result=bool)
    def setPassword(self, password):
        assert self._use_single_password
        assert password
        if not self._update_password_for_directory_and_unlock_wallets(old_password=self._password, new_password=password):
            return False
        self._password = password
        return True

    def _update_password_for_directory_and_unlock_wallets(self, *, old_password, new_password):
        # note: this assumes all wallet files are in a single directory.
        # change wallet passwords:
        ret = self.daemon.update_password_for_directory(old_password=old_password, new_password=new_password)
        # If some wallets just had their password changed, they got "locked" by wallet.update_password().
        # If the password is not unified yet, other loaded wallets might still be unlocked.
        # restore the invariant that all loaded wallets in qml must be unlocked:
        for w in self.daemon.get_wallets().values():
            if not w.is_unlocked():
                w.unlock(new_password)
            assert w.is_unlocked()
        return ret

    @pyqtProperty(QENewWalletWizard, notify=newWalletWizardChanged)
    def newWalletWizard(self):
        if not self._new_wallet_wizard:
            self._new_wallet_wizard = QENewWalletWizard(self, self.plugins)

        return self._new_wallet_wizard

    @pyqtProperty(QEServerConnectWizard, notify=serverConnectWizardChanged)
    def serverConnectWizard(self):
        if not self._server_connect_wizard:
            self._server_connect_wizard = QEServerConnectWizard(self)

        return self._server_connect_wizard

    @pyqtProperty(QETermsOfUseWizard, notify=termsOfUseWizardChanged)
    def termsOfUseWizard(self):
        if not self._terms_of_use_wizard:
            self._terms_of_use_wizard = QETermsOfUseWizard(self)
        return self._terms_of_use_wizard

    @pyqtSlot()
    def startNetwork(self):
        self.daemon.start_network()

    @pyqtSlot(str, str, str, result=bool)
    def verifyMessage(self, address, message, signature):
        address = address.strip()
        message = message.strip().encode('utf-8')
        if not is_address(address):
            return False
        try:
            # This can throw on invalid base64
            sig = base64.b64decode(str(signature.strip()), validate=True)
            verified = verify_usermessage_with_address(address, sig, message)
        except Exception as e:
            verified = False
        return verified

    @pyqtSlot(str, result=int)
    def passwordStrength(self, password):
        if len(password) == 0:
            return 0
        return check_password_strength(password)[0]