bitcoinafterlife/bal-electrum-plugin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity

Compare commits

base: bitcoinafterlife:v0.1.0
Branches Tags
bitcoinafterlife:main
bitcoinafterlife:refactoring
bitcoinafterlife:0.1.1
bitcoinafterlife:v0.2.0
bitcoinafterlife:v0.1.0
...
compare: bitcoinafterlife:main
Branches Tags
bitcoinafterlife:main
bitcoinafterlife:refactoring
bitcoinafterlife:0.1.1
bitcoinafterlife:v0.2.0
bitcoinafterlife:v0.1.0

No commits in common. "v0.1.0" and "main" have entirely different histories.
v0.1.0 ... main

15 changed files with 4022 additions and 1519 deletions
Show Stats Expand all files Collapse all files

1
VERSION Normal file
View File

@ -0,0 +1 @@
0.2.2b

19
__init__.py
View File

@ -1,20 +1 @@
from electrum.i18n import _
import subprocess
from . import bal_resources
BUILD_NUMBER = 0
REVISION_NUMBER = 1
VERSION_NUMBER = 0
def _version():
return f'{VERSION_NUMBER}.{REVISION_NUMBER}-{BUILD_NUMBER}'
version = _version()
author = "Bal Enterprise inc."
fullname = _('B.A.L.')
description = ''.join([
"<img src='",bal_resources.icon_path('bal16x16.png'),"'>", _("Bitcoin After Life"), '<br/>',
_("For more information, visit"),
" <a href=\"https://bitcoin-after.life/\">https://bitcoin-after.life/</a><br/>",
"<p style='font-size:8pt;vertialAlign:bottom'>Version: ", _version(),"</p>"
])
#available_for = ['qt', 'cmdline', 'qml']
available_for = ['qt']

185
bal.py
View File

@ -1,132 +1,149 @@
import random import random
import os import os
from hashlib import sha256 import zipfile as zipfile_lib
from typing import NamedTuple, Optional, Dict, Tuple
from electrum.plugin import BasePlugin from electrum.plugin import BasePlugin
from electrum.util import to_bytes, bfh
from electrum import json_db from electrum import json_db
from electrum.transaction import tx_from_any from electrum.transaction import tx_from_any
from . import util as Util
from . import willexecutors as Willexecutors
import os import os
def get_will_settings(x):
print(x)
json_db.register_dict('heirs', tuple, None) json_db.register_dict('heirs', tuple, None)
json_db.register_dict('will', lambda x: get_will(x), None) json_db.register_dict('will', dict,None)
json_db.register_dict('will_settings', lambda x:x, None) json_db.register_dict('will_settings', lambda x:x,None)
#{'rubiconda': ['bcrt1qgv0wu4v6kjzef5mnxfh2m9z6y7mez0ja0tt8mu', '45%', '1y'], 'veronica': ['bcrt1q6vxuvwrt8x5c9u9u29y5uq7frscr0vgc2dy60j', '15%', '1y']}
from electrum.logging import get_logger from electrum.logging import get_logger
def get_will_settings(x):
print(x)
def get_will(x): def get_will(x):
try: try:
#print("______________________________________________________________________________________________________")
#print(x)
x['tx']=tx_from_any(x['tx']) x['tx']=tx_from_any(x['tx'])
except Exception as e: except Exception as e:
#Util.print_var(x)
raise e raise e
return x return x
class BalConfig():
def __init__(self, config, name, default):
print("init bal_config")
self.config = config
self.name = name
self.default = default
def get(self,default=None):
v = self.config.get(self.name, default)
if v is None:
if not default is None:
v = default
else:
v = self.default
return v
def set(self,value,save=True):
self.config.set_key(self.name,value,save=save)
class BalPlugin(BasePlugin): class BalPlugin(BasePlugin):
LOCKTIME_TIME = "bal_locktime_time"
LOCKTIME_BLOCKS = "bal_locktime_blocks"
LOCKTIMEDELTA_TIME = "bal_locktimedelta_time"
LOCKTIMEDELTA_BLOCKS = "bal_locktimedelta_blocks"
TX_FEES = "bal_tx_fees"
BROADCAST = "bal_broadcast"
ASK_BROADCAST = "bal_ask_broadcast"
INVALIDATE = "bal_invalidate"
ASK_INVALIDATE = "bal_ask_invalidate"
PREVIEW = "bal_preview"
SAVE_TXS = "bal_save_txs"
WILLEXECUTORS = "bal_willexecutors"
PING_WILLEXECUTORS = "bal_ping_willexecutors"
ASK_PING_WILLEXECUTORS = "bal_ask_ping_willexecutors"
NO_WILLEXECUTOR = "bal_no_willexecutor"
HIDE_REPLACED = "bal_hide_replaced"
HIDE_INVALIDATED = "bal_hide_invalidated"
ALLOW_REPUSH = "bal_allow_repush"
DEFAULT_SETTINGS={
LOCKTIME_TIME: 90,
LOCKTIME_BLOCKS: 144*90,
LOCKTIMEDELTA_TIME: 7,
LOCKTIMEDELTA_BLOCKS:144*7,
TX_FEES: 100,
BROADCAST: True,
ASK_BROADCAST: True,
INVALIDATE: True,
ASK_INVALIDATE: True,
PREVIEW: True,
SAVE_TXS: True,
PING_WILLEXECUTORS: False,
ASK_PING_WILLEXECUTORS: False,
NO_WILLEXECUTOR: False,
HIDE_REPLACED:True,
HIDE_INVALIDATED:True,
ALLOW_REPUSH: False,
WILLEXECUTORS: {
'http://bitcoin-after.life:9137': {
"base_fee": 100000,
"status": "New",
"info":"Bitcoin After Life Will Executor",
"address":"bcrt1qa5cntu4hgadw8zd3n6sq2nzjy34sxdtd9u0gp7",
"selected":True
}
},
}
LATEST_VERSION = '1' LATEST_VERSION = '1'
KNOWN_VERSIONS = ('0', '1') KNOWN_VERSIONS = ('0', '1')
assert LATEST_VERSION in KNOWN_VERSIONS assert LATEST_VERSION in KNOWN_VERSIONS
def version():
try:
f=""
with open("VERSION","r") as f:
f = str(f.readline())
return f
except:
return "unknown"
SIZE = (159, 97) SIZE = (159, 97)
def __init__(self, parent, config, name): def __init__(self, parent, config, name):
self.logger= get_logger(__name__) print("init bal_plugin")
self.logger = get_logger(__name__)
BasePlugin.__init__(self, parent, config, name) BasePlugin.__init__(self, parent, config, name)
self.base_dir = os.path.join(config.electrum_path(), 'bal') self.base_dir = os.path.join(config.electrum_path(), 'bal')
self.logger.info(self.base_dir) self.plugin_dir = os.path.split(os.path.realpath(__file__))[0]
zipfile="/".join(self.plugin_dir.split("/")[:-1])
#print("real path",os.path.realpath(__file__))
#self.logger.info(self.base_dir)
#print("base_dir:", self.base_dir)
#print("suca:",zipfile)
#print("plugin_dir:", self.plugin_dir)
import sys
sys.path.insert(0, zipfile)
#print("sono state listate?")
self.parent = parent self.parent = parent
self.config = config self.config = config
self.name = name self.name = name
self._hide_invalidated= self.config_get(self.HIDE_INVALIDATED)
self._hide_replaced= self.config_get(self.HIDE_REPLACED) self.ASK_BROADCAST = BalConfig(config, "bal_ask_broadcast", True)
self.plugin_dir = os.path.split(os.path.realpath(__file__))[0] self.BROADCAST = BalConfig(config, "bal_broadcast", True)
self.LOCKTIME_TIME = BalConfig(config, "bal_locktime_time", 90)
self.LOCKTIME_BLOCKS = BalConfig(config, "bal_locktime_blocks", 144*90)
self.LOCKTIMEDELTA_TIME = BalConfig(config, "bal_locktimedelta_time", 7)
self.LOCKTIMEDELTA_BLOCKS = BalConfig(config, "bal_locktimedelta_blocks", 144*7)
self.ENABLE_MULTIVERSE = BalConfig(config, "bal_enable_multiverse", False)
self.TX_FEES = BalConfig(config, "bal_tx_fees", 100)
self.INVALIDATE = BalConfig(config, "bal_invalidate", True)
self.ASK_INVALIDATE = BalConfig(config, "bal_ask_invalidate", True)
self.PREVIEW = BalConfig(config, "bal_preview", True)
self.SAVE_TXS = BalConfig(config, "bal_save_txs", True)
self.WILLEXECUTORS = BalConfig(config, "bal_willexecutors", True)
self.PING_WILLEXECUTORS = BalConfig(config, "bal_ping_willexecutors", True)
self.ASK_PING_WILLEXECUTORS = BalConfig(config, "bal_ask_ping_willexecutors", True)
self.NO_WILLEXECUTOR = BalConfig(config, "bal_no_willexecutor", True)
self.HIDE_REPLACED = BalConfig(config, "bal_hide_replaced", True)
self.HIDE_INVALIDATED = BalConfig(config, "bal_hide_invalidated", True)
self.ALLOW_REPUSH = BalConfig(config, "bal_allow_repush", True)
self.FIRST_EXECUTION = BalConfig(config, "bal_first_execution", True)
self.WILLEXECUTORS = BalConfig(config, "bal_willexecutors", {
"mainnet": {
'https://we.bitcoin-after.life': {
"base_fee": 100000,
"status": "New",
"info":"Bitcoin After Life Will Executor",
"address":"bcrt1qa5cntu4hgadw8zd3n6sq2nzjy34sxdtd9u0gp7",
"selected":True
}
}
})
self.WILL_SETTINGS = BalConfig(config, "bal_will_settings", {
'baltx_fees':100,
'threshold':'180d',
'locktime':'1y',
})
self._hide_invalidated= self.HIDE_INVALIDATED.get()
self._hide_replaced= self.HIDE_REPLACED.get()
def resource_path(self,*parts): def resource_path(self,*parts):
return os.path.join(self.plugin_dir, *parts) return os.path.join(self.plugin_dir, *parts)
def config_get(self,key):
v = self.config.get(key,None)
if v is None:
self.config.set_key(key,self.DEFAULT_SETTINGS[key],save=True)
v = self.DEFAULT_SETTINGS[key]
return v
def hide_invalidated(self): def hide_invalidated(self):
self._hide_invalidated = not self._hide_invalidated self._hide_invalidated = not self._hide_invalidated
self.config.set_key(BalPlugin.HIDE_INVALIDATED,self.hide_invalidated,save=True) self.HIDE_INVALIDATED.set(self._hide_invalidated)
def hide_replaced(self): def hide_replaced(self):
self._hide_replaced = not self._hide_replaced self._hide_replaced = not self._hide_replaced
self.config.set_key(BalPlugin.HIDE_REPLACED,self.hide_invalidated,save=True) self.HIDE_REPLACED.set(self._hide_replaced)
def default_will_settings(self):
return {
'tx_fees':100,
'threshold':'180d',
'locktime':'1y',
}
def validate_will_settings(self,will_settings): def validate_will_settings(self,will_settings):
if int(will_settings.get('tx_fees',1))<1: print(type(will_settings))
will_settings['tx_fees']=1 print(will_settings.get('baltx_fees',1),1)
if int(will_settings.get('baltx_fees',1))<1:
will_settings['baltx_fees']=1
if not will_settings.get('threshold'): if not will_settings.get('threshold'):
will_settings['threshold']='180d' will_settings['threshold']='180d'
if not will_settings.get('locktime')=='': if not will_settings.get('locktime')=='':
will_settings['locktime']='1y' will_settings['locktime']='1y'
return will_settings return will_settings
def default_will_settings(self):
return {
'baltx_fees':100,
'threshold':'180d',
'locktime':'1y'
}

2
bal_resources.py
View File

@ -2,7 +2,7 @@ import os
PLUGIN_DIR = os.path.split(os.path.realpath(__file__))[0] PLUGIN_DIR = os.path.split(os.path.realpath(__file__))[0]
DEFAULT_ICON = 'bal32x32.png' DEFAULT_ICON = 'bal32x32.png'
DEFAULT_ICON_PATH = 'icons' DEFAULT_ICON_PATH = ''
def icon_path(icon_basename: str = DEFAULT_ICON): def icon_path(icon_basename: str = DEFAULT_ICON):

21
balqt/baldialog.py
View File

@ -95,25 +95,8 @@ class bal_checkbox(QCheckBox):
def __init__(self, plugin,variable,window=None): def __init__(self, plugin,variable,window=None):
QCheckBox.__init__(self) QCheckBox.__init__(self)
self.setChecked(plugin.config_get(variable)) self.setChecked(plugin.config_get(variable))
window=window
def on_check(v): def on_check(v):
plugin.config.set_key(variable, v == Qt.CheckState.Checked, save=True) plugin.config.set_key(variable, v == 2)
if window: plugin.config_get(variable)
plugin._hide_invalidated= plugin.config_get(plugin.HIDE_INVALIDATED)
plugin._hide_replaced= plugin.config_get(plugin.HIDE_REPLACED)
window.update_all()
self.stateChanged.connect(on_check) self.stateChanged.connect(on_check)
#TODO IMPLEMENT PREVIEW DIALOG
#tx list display txid, willexecutor, qrcode, button to sign
# :def preview_dialog(self, txs):
def preview_dialog(self, txs):
w=PreviewDialog(self,txs)
w.exec()
return w
def add_info_from_will(self,tx):
for input in tx.inputs():
pass

2
balqt/closedialog.py
View File

@ -225,7 +225,7 @@ class BalCloseDialog(BalDialog):
#self._stopping=True #self._stopping=True
#self.on_success_phase2() #self.on_success_phase2()
# return # return
_logger.debug("have to sign",self.have_to_sign) _logger.debug("have to sign {}".format(self.have_to_sign))
password=None password=None
if self.have_to_sign is None: if self.have_to_sign is None:
self.msg_set_invalidating() self.msg_set_invalidating()

1
balqt/willexecutor_dialog.py
View File

@ -199,7 +199,6 @@ class WillExecutorDialog(BalDialog,MessageBoxMixin):
def __init__(self, bal_window): def __init__(self, bal_window):
BalDialog.__init__(self,bal_window.window) BalDialog.__init__(self,bal_window.window)
self.bal_plugin = bal_window.bal_plugin self.bal_plugin = bal_window.bal_plugin
self.gui_object = self.bal_plugin.gui_object
self.config = self.bal_plugin.config self.config = self.bal_plugin.config
self.window = bal_window.window self.window = bal_window.window
self.bal_window = bal_window self.bal_window = bal_window

25
heirs.py
View File

@ -14,9 +14,9 @@ from electrum.transaction import PartialTxInput, PartialTxOutput,TxOutpoint,Part
import datetime import datetime
import urllib.request import urllib.request
import urllib.parse import urllib.parse
from .bal import BalPlugin import random
from . import util as Util from .util import Util
from . import willexecutors as Willexecutors from .willexecutors import Willexecutors
if TYPE_CHECKING: if TYPE_CHECKING:
from .wallet_db import WalletDB from .wallet_db import WalletDB
from .simple_config import SimpleConfig from .simple_config import SimpleConfig
@ -38,6 +38,7 @@ def reduce_outputs(in_amount, out_amount, fee, outputs):
for output in outputs: for output in outputs:
output.value = math.floor((in_amount-fee)/out_amount * output.value) output.value = math.floor((in_amount-fee)/out_amount * output.value)
"""
#TODO: put this method inside wallet.db to replace or complete get_locktime_for_new_transaction #TODO: put this method inside wallet.db to replace or complete get_locktime_for_new_transaction
def get_current_height(network:'Network'): def get_current_height(network:'Network'):
#if no network or not up to date, just set locktime to zero #if no network or not up to date, just set locktime to zero
@ -57,7 +58,7 @@ def get_current_height(network:'Network'):
# discourage "fee sniping" # discourage "fee sniping"
height = min(chain_height, server_height) height = min(chain_height, server_height)
return height return height
"""
@ -112,7 +113,9 @@ def prepare_transactions(locktimes, available_utxos, fees, wallet):
change = get_change_output(wallet, in_amount, out_amount, fee) change = get_change_output(wallet, in_amount, out_amount, fee)
if change: if change:
outputs.append(change) outputs.append(change)
for i in range(0,100):
random.shuffle(outputs)
print(outputs)
tx = PartialTransaction.from_io(used_utxos, outputs, locktime=Util.parse_locktime_string(locktime,wallet), version=2) tx = PartialTransaction.from_io(used_utxos, outputs, locktime=Util.parse_locktime_string(locktime,wallet), version=2)
if len(description)>0: tx.description = description[:-1] if len(description)>0: tx.description = description[:-1]
else: tx.description = "" else: tx.description = ""
@ -385,8 +388,8 @@ class Heirs(dict, Logger):
if not utxos: if not utxos:
utxos = wallet.get_utxos() utxos = wallet.get_utxos()
willexecutors = Willexecutors.get_willexecutors(bal_plugin) or {} willexecutors = Willexecutors.get_willexecutors(bal_plugin) or {}
self.decimal_point=bal_plugin.config.get_decimal_point() self.decimal_point=bal_plugin.get_decimal_point()
no_willexecutors = bal_plugin.config_get(BalPlugin.NO_WILLEXECUTOR) no_willexecutors = bal_plugin.NO_WILLEXECUTOR.get()
for utxo in utxos: for utxo in utxos:
if utxo.value_sats()> 0*tx_fees: if utxo.value_sats()> 0*tx_fees:
balance += utxo.value_sats() balance += utxo.value_sats()
@ -563,10 +566,7 @@ class Heirs(dict, Logger):
def validate_amount(amount): def validate_amount(amount):
try: try:
if Util.is_perc(amount): famount = float(amount[:-1]) if Util.is_perc(amount) else float(amount)
famount = float(amount[:-1])
else:
famount = float(amount)
if famount <= 0.00000001: if famount <= 0.00000001:
raise AmountNotValid(f"amount have to be positive {famount} < 0") raise AmountNotValid(f"amount have to be positive {famount} < 0")
except Exception as e: except Exception as e:
@ -575,9 +575,8 @@ class Heirs(dict, Logger):
def validate_locktime(locktime,timestamp_to_check=False): def validate_locktime(locktime,timestamp_to_check=False):
try: try:
locktime = Util.parse_locktime_string(locktime,None)
if timestamp_to_check: if timestamp_to_check:
if locktime < timestamp_to_check: if Util.parse_locktime_string(locktime,None) < timestamp_to_check:
raise HeirExpiredException() raise HeirExpiredException()
except Exception as e: except Exception as e:
raise LocktimeNotValid(f"locktime string not properly formatted, {e}") raise LocktimeNotValid(f"locktime string not properly formatted, {e}")

9
manifest.json Normal file
View File

@ -0,0 +1,9 @@
{
"name": "BAL",
"fullname": "Bitcoin After Life",
"description": "Provides free and decentralized inheritance support<br> Version: 0.2.2b",
"author":"Svatantrya",
"available_for": ["qt"],
"icon":"icons/bal32x32.png"
}

2705
qt.py
View File

File diff suppressed because it is too large Load Diff

809
util.py
View File

@ -8,451 +8,430 @@ import urllib.parse
from electrum.util import write_json_file,FileImportFailed,FileExportFailed from electrum.util import write_json_file,FileImportFailed,FileExportFailed
LOCKTIME_THRESHOLD = 500000000 LOCKTIME_THRESHOLD = 500000000
def locktime_to_str(locktime): class Util:
try: def locktime_to_str(locktime):
locktime=int(locktime)
if locktime > LOCKTIME_THRESHOLD:
dt = datetime.fromtimestamp(locktime).isoformat()
return dt
except Exception as e:
#print(e)
pass
return str(locktime)
def str_to_locktime(locktime):
try:
if locktime[-1] in ('y','d','b'):
return locktime
else: return int(locktime)
except Exception as e:
pass
#print(e)
dt_object = datetime.fromisoformat(locktime)
timestamp = dt_object.timestamp()
return int(timestamp)
def parse_locktime_string(locktime,w=None):
try:
return int(locktime)
except Exception as e:
pass
#print("parse_locktime_string",e)
try:
now = datetime.now()
if locktime[-1] == 'y':
locktime = str(int(locktime[:-1])*365) + "d"
if locktime[-1] == 'd':
return int((now + timedelta(days = int(locktime[:-1]))).replace(hour=0,minute=0,second=0,microsecond=0).timestamp())
if locktime[-1] == 'b':
locktime = int(locktime[:-1])
height = 0
if w:
height = get_current_height(w.network)
locktime+=int(height)
return int(locktime)
except Exception as e:
print("parse_locktime_string",e)
#raise e
return 0
def int_locktime(seconds=0,minutes=0,hours=0, days=0, blocks = 0):
return int(seconds + minutes*60 + hours*60*60 + days*60*60*24 + blocks * 600)
def encode_amount(amount, decimal_point):
if is_perc(amount):
return amount
else:
try: try:
return int(float(amount)*pow(10,decimal_point)) locktime=int(locktime)
except: if locktime > LOCKTIME_THRESHOLD:
return 0 dt = datetime.fromtimestamp(locktime).isoformat()
return dt
def decode_amount(amount,decimal_point): except Exception as e:
if is_perc(amount): pass
return amount return str(locktime)
else:
num=8-decimal_point
basestr="{{:0{}.{}f}}".format(num,num)
return "{:08.8f}".format(float(amount)/pow(10,decimal_point))
def is_perc(value): def str_to_locktime(locktime):
try:
if locktime[-1] in ('y','d','b'):
return locktime
else: return int(locktime)
except Exception as e:
pass
dt_object = datetime.fromisoformat(locktime)
timestamp = dt_object.timestamp()
return int(timestamp)
def parse_locktime_string(locktime,w=None):
try:
return int(locktime)
except Exception as e:
pass
try:
now = datetime.now()
if locktime[-1] == 'y':
locktime = str(int(locktime[:-1])*365) + "d"
if locktime[-1] == 'd':
return int((now + timedelta(days = int(locktime[:-1]))).replace(hour=0,minute=0,second=0,microsecond=0).timestamp())
if locktime[-1] == 'b':
locktime = int(locktime[:-1])
height = 0
if w:
height = Util.get_current_height(w.network)
locktime+=int(height)
return int(locktime)
except Exception as e:
pass
return 0
def int_locktime(seconds=0,minutes=0,hours=0, days=0, blocks = 0):
return int(seconds + minutes*60 + hours*60*60 + days*60*60*24 + blocks * 600)
def encode_amount(amount, decimal_point):
if Util.is_perc(amount):
return amount
else:
try:
return int(float(amount)*pow(10,decimal_point))
except:
return 0
def decode_amount(amount,decimal_point):
if Util.is_perc(amount):
return amount
else:
num=8-decimal_point
basestr="{{:0{}.{}f}}".format(num,num)
return "{:08.8f}".format(float(amount)/pow(10,decimal_point))
def is_perc(value):
try: try:
return value[-1] == '%' return value[-1] == '%'
except: except:
return False return False
def cmp_array(heira,heirb): def cmp_array(heira,heirb):
try: try:
if not len(heira) == len(heirb): if not len(heira) == len(heirb):
return False
for h in range(0,len(heira)):
if not heira[h] == heirb[h]:
return False return False
return True for h in range(0,len(heira)):
except: if not heira[h] == heirb[h]:
return False return False
def cmp_heir(heira,heirb):
if heira[0] == heirb[0] and heira[1] == heirb[1]:
return True
return False
def cmp_willexecutor(willexecutora,willexecutorb):
if willexecutora == willexecutorb:
return True
try:
if willexecutora['url']==willexecutorb['url'] and willexecutora['address'] == willexecutorb['address'] and willexecutora['base_fee']==willexecutorb['base_fee']:
return True return True
except: except:
return False
return False
def search_heir_by_values(heirs,heir,values):
#print()
for h,v in heirs.items():
found = False
for val in values:
if val in v and v[val] != heir[val]:
found = True
if not found:
return h
return False
def cmp_heir_by_values(heira,heirb,values):
for v in values:
if heira[v] != heirb[v]:
return False return False
return True
def cmp_heirs_by_values(heirsa,heirsb,values,exclude_willexecutors=False,reverse = True): def cmp_heir(heira,heirb):
for heira in heirsa: if heira[0] == heirb[0] and heira[1] == heirb[1]:
if (exclude_willexecutors and not "w!ll3x3c\"" in heira) or not exclude_willexecutors: return True
return False
def cmp_willexecutor(willexecutora,willexecutorb):
if willexecutora == willexecutorb:
return True
try:
if willexecutora['url']==willexecutorb['url'] and willexecutora['address'] == willexecutorb['address'] and willexecutora['base_fee']==willexecutorb['base_fee']:
return True
except:
return False
return False
def search_heir_by_values(heirs,heir,values):
for h,v in heirs.items():
found = False found = False
for heirb in heirsb: for val in values:
if cmp_heir_by_values(heirsa[heira],heirsb[heirb],values): if val in v and v[val] != heir[val]:
found=True found = True
if not found: if not found:
#print(f"not_found {heira}--{heirsa[heira]}") return h
return False
def cmp_heir_by_values(heira,heirb,values):
for v in values:
if heira[v] != heirb[v]:
return False return False
if reverse:
return cmp_heirs_by_values(heirsb,heirsa,values,exclude_willexecutors=exclude_willexecutors,reverse=False)
else:
return True return True
def cmp_heirs(heirsa,heirsb,cmp_function = lambda x,y: x[0]==y[0] and x[3]==y[3],reverse=True): def cmp_heirs_by_values(heirsa,heirsb,values,exclude_willexecutors=False,reverse = True):
try: for heira in heirsa:
for heir in heirsa: if (exclude_willexecutors and not "w!ll3x3c\"" in heira) or not exclude_willexecutors:
if not "w!ll3x3c\"" in heir: found = False
if not heir in heirsb or not cmp_function(heirsa[heir],heirsb[heir]): for heirb in heirsb:
if not search_heir_by_values(heirsb,heirsa[heir],[0,3]): if Util.cmp_heir_by_values(heirsa[heira],heirsb[heirb],values):
return False found=True
if not found:
return False
if reverse: if reverse:
return cmp_heirs(heirsb,heirsa,cmp_function,False) return Util.cmp_heirs_by_values(heirsb,heirsa,values,exclude_willexecutors=exclude_willexecutors,reverse=False)
else: else:
return True return True
except Exception as e:
raise e
return False
def cmp_inputs(inputsa,inputsb): def cmp_heirs(heirsa,heirsb,cmp_function = lambda x,y: x[0]==y[0] and x[3]==y[3],reverse=True):
if len(inputsa) != len(inputsb): try:
return False for heir in heirsa:
for inputa in inputsa: if not "w!ll3x3c\"" in heir:
if not in_utxo(inputa,inputsb): if not heir in heirsb or not cmp_function(heirsa[heir],heirsb[heir]):
return False if not Util.search_heir_by_values(heirsb,heirsa[heir],[0,3]):
return True return False
if reverse:
def cmp_outputs(outputsa,outputsb,willexecutor_output = None): return Util.cmp_heirs(heirsb,heirsa,cmp_function,False)
if len(outputsa) != len(outputsb):
return False
for outputa in outputsa:
if not cmp_output(outputa,willexecutor_output):
if not in_output(outputa,outputsb):
return False
return True
def cmp_txs(txa,txb):
if not cmp_inputs(txa.inputs(),txb.inputs()):
return False
if not cmp_outputs(txa.outputs(),txb.outputs()):
return False
return True
def get_value_amount(txa,txb):
outputsa=txa.outputs()
outputsb=txb.outputs()
value_amount = 0
#if len(outputsa) != len(outputsb):
# print("outputlen is different")
# return False
for outa in outputsa:
same_amount,same_address = in_output(outa,txb.outputs())
if not (same_amount or same_address):
#print("outa notin txb", same_amount,same_address)
return False
if same_amount and same_address:
value_amount+=outa.value
if same_amount:
pass
#print("same amount")
if same_address:
pass
#print("same address")
return value_amount
#not needed
#for outb in outputsb:
# if not in_output(outb,txa.outputs()):
# print("outb notin txb")
# return False
def chk_locktime(timestamp_to_check,block_height_to_check,locktime):
#TODO BUG: WHAT HAPPEN AT THRESHOLD?
locktime=int(locktime)
if locktime > LOCKTIME_THRESHOLD and locktime > timestamp_to_check:
return True
elif locktime < LOCKTIME_THRESHOLD and locktime > block_height_to_check:
return True
else:
return False
def anticipate_locktime(locktime,blocks=0,hours=0,days=0):
locktime = int(locktime)
out=0
if locktime> LOCKTIME_THRESHOLD:
seconds = blocks*600 + hours*3600 + days*86400
dt = datetime.fromtimestamp(locktime)
dt -= timedelta(seconds=seconds)
out = dt.timestamp()
else:
blocks -= hours*6 + days*144
out = locktime + blocks
if out < 1:
out = 1
return out
def cmp_locktime(locktimea,locktimeb):
if locktimea==locktimeb:
return 0
strlocktime = str(locktimea)
strlocktimeb = str(locktimeb)
intlocktimea = str_to_locktime(strlocktimea)
intlocktimeb = str_to_locktime(strlocktimeb)
if locktimea[-1] in "ydb":
if locktimeb[-1] == locktimea[-1]:
return int(strlocktimea[-1])-int(strlocktimeb[-1])
else:
return int(locktimea)-(locktimeb)
def get_lowest_valid_tx(available_utxos,will):
will = sorted(will.items(),key = lambda x: x[1]['tx'].locktime)
for txid,willitem in will.items():
pass
def get_locktimes(will):
locktimes = {}
for txid,willitem in will.items():
locktimes[willitem['tx'].locktime]=True
return locktimes.keys()
def get_lowest_locktimes(locktimes):
sorted_timestamp=[]
sorted_block=[]
for l in locktimes:
#print("locktime:",parse_locktime_string(l))
l=parse_locktime_string(l)
if l < LOCKTIME_THRESHOLD:
bisect.insort(sorted_block,l)
else:
bisect.insort(sorted_timestamp,l)
return sorted(sorted_timestamp), sorted(sorted_block)
def get_lowest_locktimes_from_will(will):
return get_lowest_locktimes(get_locktimes(will))
def search_willtx_per_io(will,tx):
for wid, w in will.items():
if cmp_txs(w['tx'],tx['tx']):
return wid,w
return None, None
def invalidate_will(will):
raise Exception("not implemented")
def get_will_spent_utxos(will):
utxos=[]
for txid,willitem in will.items():
utxos+=willitem['tx'].inputs()
return utxos
def utxo_to_str(utxo):
try: return utxo.to_str()
except Exception as e: pass
try: return utxo.prevout.to_str()
except Exception as e: pass
return str(utxo)
def cmp_utxo(utxoa,utxob):
utxoa=utxo_to_str(utxoa)
utxob=utxo_to_str(utxob)
if utxoa == utxob:
#if utxoa.prevout.txid==utxob.prevout.txid and utxoa.prevout.out_idx == utxob.prevout.out_idx:
return True
else:
return False
def in_utxo(utxo, utxos):
for s_u in utxos:
if cmp_utxo(s_u,utxo):
return True
return False
def txid_in_utxo(txid,utxos):
for s_u in utxos:
if s_u.prevout.txid == txid:
return True
return False
def cmp_output(outputa,outputb):
return outputa.address == outputb.address and outputa.value == outputb.value
def in_output(output,outputs):
for s_o in outputs:
if cmp_output(s_o,output):
return True
return False
#check all output with the same amount if none have the same address it can be a change
#return true true same address same amount
#return true false same amount different address
#return false false different amount, different address not found
def din_output(out,outputs):
same_amount=[]
for s_o in outputs:
if int(out.value) == int(s_o.value):
same_amount.append(s_o)
if out.address==s_o.address:
#print("SAME_:",out.address,s_o.address)
return True, True
else: else:
return True
except Exception as e:
raise e
return False
def cmp_inputs(inputsa,inputsb):
if len(inputsa) != len(inputsb):
return False
for inputa in inputsa:
if not Util.in_utxo(inputa,inputsb):
return False
return True
def cmp_outputs(outputsa,outputsb,willexecutor_output = None):
if len(outputsa) != len(outputsb):
return False
for outputa in outputsa:
if not Util.cmp_output(outputa,willexecutor_output):
if not Util.in_output(outputa,outputsb):
return False
return True
def cmp_txs(txa,txb):
if not Util.cmp_inputs(txa.inputs(),txb.inputs()):
return False
if not Util.cmp_outputs(txa.outputs(),txb.outputs()):
return False
return True
def get_value_amount(txa,txb):
outputsa=txa.outputs()
outputsb=txb.outputs()
value_amount = 0
for outa in outputsa:
same_amount,same_address = Util.in_output(outa,txb.outputs())
if not (same_amount or same_address):
return False
if same_amount and same_address:
value_amount+=outa.value
if same_amount:
pass
if same_address:
pass pass
#print("NOT SAME_:",out.address,s_o.address)
if len(same_amount)>0: return value_amount
return True, False
else:return False, False
def get_change_output(wallet,in_amount,out_amount,fee):
change_amount = int(in_amount - out_amount - fee) def chk_locktime(timestamp_to_check,block_height_to_check,locktime):
if change_amount > wallet.dust_threshold(): #TODO BUG: WHAT HAPPEN AT THRESHOLD?
change_addresses = wallet.get_change_addresses_for_new_transaction() locktime=int(locktime)
out = PartialTxOutput.from_address_and_value(change_addresses[0], change_amount) if locktime > LOCKTIME_THRESHOLD and locktime > timestamp_to_check:
out.is_change = True return True
elif locktime < LOCKTIME_THRESHOLD and locktime > block_height_to_check:
return True
else:
return False
def anticipate_locktime(locktime,blocks=0,hours=0,days=0):
locktime = int(locktime)
out=0
if locktime> LOCKTIME_THRESHOLD:
seconds = blocks*600 + hours*3600 + days*86400
dt = datetime.fromtimestamp(locktime)
dt -= timedelta(seconds=seconds)
out = dt.timestamp()
else:
blocks -= hours*6 + days*144
out = locktime + blocks
if out < 1:
out = 1
return out return out
def cmp_locktime(locktimea,locktimeb):
def get_current_height(network:'Network'): if locktimea==locktimeb:
#if no network or not up to date, just set locktime to zero return 0
if not network: strlocktime = str(locktimea)
return 0 strlocktimeb = str(locktimeb)
chain = network.blockchain() intlocktimea = Util.str_to_locktime(strlocktimea)
if chain.is_tip_stale(): intlocktimeb = Util.str_to_locktime(strlocktimeb)
return 0 if locktimea[-1] in "ydb":
# figure out current block height if locktimeb[-1] == locktimea[-1]:
chain_height = chain.height() # learnt from all connected servers, SPV-checked return int(strlocktimea[-1])-int(strlocktimeb[-1])
server_height = network.get_server_height() # height claimed by main server, unverified else:
# note: main server might be lagging (either is slow, is malicious, or there is an SPV-invisible-hard-fork) return int(locktimea)-(locktimeb)
# - if it's lagging too much, it is the network's job to switch away
if server_height < chain_height - 10:
# the diff is suspiciously large... give up and use something non-fingerprintable
return 0
# discourage "fee sniping"
height = min(chain_height, server_height)
return height
def print_var(var,name = "",veryverbose=False): def get_lowest_valid_tx(available_utxos,will):
print(f"---{name}---") will = sorted(will.items(),key = lambda x: x[1]['tx'].locktime)
if not var is None: for txid,willitem in will.items():
pass
def get_locktimes(will):
locktimes = {}
for txid,willitem in will.items():
locktimes[willitem['tx'].locktime]=True
return locktimes.keys()
def get_lowest_locktimes(locktimes):
sorted_timestamp=[]
sorted_block=[]
for l in locktimes:
l=Util.parse_locktime_string(l)
if l < LOCKTIME_THRESHOLD:
bisect.insort(sorted_block,l)
else:
bisect.insort(sorted_timestamp,l)
return sorted(sorted_timestamp), sorted(sorted_block)
def get_lowest_locktimes_from_will(will):
return Util.get_lowest_locktimes(Util.get_locktimes(will))
def search_willtx_per_io(will,tx):
for wid, w in will.items():
if Util.cmp_txs(w['tx'],tx['tx']):
return wid,w
return None, None
def invalidate_will(will):
raise Exception("not implemented")
def get_will_spent_utxos(will):
utxos=[]
for txid,willitem in will.items():
utxos+=willitem['tx'].inputs()
return utxos
def utxo_to_str(utxo):
try: return utxo.to_str()
except Exception as e: pass
try: return utxo.prevout.to_str()
except Exception as e: pass
return str(utxo)
def cmp_utxo(utxoa,utxob):
utxoa=Util.utxo_to_str(utxoa)
utxob=Util.utxo_to_str(utxob)
if utxoa == utxob:
return True
else:
return False
def in_utxo(utxo, utxos):
for s_u in utxos:
if Util.cmp_utxo(s_u,utxo):
return True
return False
def txid_in_utxo(txid,utxos):
for s_u in utxos:
if s_u.prevout.txid == txid:
return True
return False
def cmp_output(outputa,outputb):
return outputa.address == outputb.address and outputa.value == outputb.value
def in_output(output,outputs):
for s_o in outputs:
if Util.cmp_output(s_o,output):
return True
return False
#check all output with the same amount if none have the same address it can be a change
#return true true same address same amount
#return true false same amount different address
#return false false different amount, different address not found
def din_output(out,outputs):
same_amount=[]
for s_o in outputs:
if int(out.value) == int(s_o.value):
same_amount.append(s_o)
if out.address==s_o.address:
return True, True
else:
pass
if len(same_amount)>0:
return True, False
else:return False, False
def get_change_output(wallet,in_amount,out_amount,fee):
change_amount = int(in_amount - out_amount - fee)
if change_amount > wallet.dust_threshold():
change_addresses = wallet.get_change_addresses_for_new_transaction()
out = PartialTxOutput.from_address_and_value(change_addresses[0], change_amount)
out.is_change = True
return out
def get_current_height(network:'Network'):
#if no network or not up to date, just set locktime to zero
if not network:
return 0
chain = network.blockchain()
if chain.is_tip_stale():
return 0
# figure out current block height
chain_height = chain.height() # learnt from all connected servers, SPV-checked
server_height = network.get_server_height() # height claimed by main server, unverified
# note: main server might be lagging (either is slow, is malicious, or there is an SPV-invisible-hard-fork)
# - if it's lagging too much, it is the network's job to switch away
if server_height < chain_height - 10:
# the diff is suspiciously large... give up and use something non-fingerprintable
return 0
# discourage "fee sniping"
height = min(chain_height, server_height)
return height
def print_var(var,name = "",veryverbose=False):
print(f"---{name}---")
if not var is None:
try:
print("doc:",doc(var))
except: pass
try:
print("str:",str(var))
except: pass
try:
print("repr",repr(var))
except:pass
try:
print("dict",dict(var))
except:pass
try:
print("dir",dir(var))
except:pass
try:
print("type",type(var))
except:pass
try:
print("to_json",var.to_json())
except: pass
try:
print("__slotnames__",var.__slotnames__)
except:pass
print(f"---end {name}---")
def print_utxo(utxo, name = ""):
print(f"---utxo-{name}---")
Util.print_var(utxo,name)
Util.print_prevout(utxo.prevout,name)
Util.print_var(utxo.script_sig,f"{name}-script-sig")
Util.print_var(utxo.witness,f"{name}-witness")
print("_TxInput__address:",utxo._TxInput__address)
print("_TxInput__scriptpubkey:",utxo._TxInput__scriptpubkey)
print("_TxInput__value_sats:",utxo._TxInput__value_sats)
print(f"---utxo-end {name}---")
def print_prevout(prevout, name = ""):
print(f"---prevout-{name}---")
Util.print_var(prevout,f"{name}-prevout")
Util.print_var(prevout._asdict())
print(f"---prevout-end {name}---")
def export_meta_gui(electrum_window: 'ElectrumWindow', title, exporter):
filter_ = "All files (*)"
filename = getSaveFileName(
parent=electrum_window,
title=_("Select file to save your {}").format(title),
filename='BALplugin_{}'.format(title),
filter=filter_,
config=electrum_window.config,
)
if not filename:
return
try: try:
print("doc:",doc(var)) exporter(filename)
except: pass except FileExportFailed as e:
try: electrum_window.show_critical(str(e))
print("str:",str(var)) else:
except: pass electrum_window.show_message(_("Your {0} were exported to '{1}'")
try: .format(title, str(filename)))
print("repr",repr(var))
except:pass
try:
print("dict",dict(var))
except:pass
try:
print("dir",dir(var))
except:pass
try:
print("type",type(var))
except:pass
try:
print("to_json",var.to_json())
except: pass
try:
print("__slotnames__",var.__slotnames__)
except:pass
print(f"---end {name}---")
def print_utxo(utxo, name = ""):
print(f"---utxo-{name}---")
print_var(utxo,name)
print_prevout(utxo.prevout,name)
print_var(utxo.script_sig,f"{name}-script-sig")
print_var(utxo.witness,f"{name}-witness")
#print("madonnamaiala_TXInput__scriptpubkey:",utxo._TXInput__scriptpubkey)
print("_TxInput__address:",utxo._TxInput__address)
print("_TxInput__scriptpubkey:",utxo._TxInput__scriptpubkey)
print("_TxInput__value_sats:",utxo._TxInput__value_sats)
print(f"---utxo-end {name}---")
def print_prevout(prevout, name = ""):
print(f"---prevout-{name}---")
print_var(prevout,f"{name}-prevout")
print_var(prevout._asdict())
print(f"---prevout-end {name}---")
def export_meta_gui(electrum_window: 'ElectrumWindow', title, exporter):
filter_ = "All files (*)"
filename = getSaveFileName(
parent=electrum_window,
title=_("Select file to save your {}").format(title),
filename='BALplugin_{}'.format(title),
filter=filter_,
config=electrum_window.config,
)
if not filename:
return
try:
exporter(filename)
except FileExportFailed as e:
electrum_window.show_critical(str(e))
else:
electrum_window.show_message(_("Your {0} were exported to '{1}'")
.format(title, str(filename)))
def copy(dicto,dictfrom): def copy(dicto,dictfrom):
for k,v in dictfrom.items(): for k,v in dictfrom.items():
dicto[k]=v dicto[k]=v

69
wallet_util/bal_wallet_utils.py Executable file
View File

@ -0,0 +1,69 @@
#!env/bin/python3
from electrum.storage import WalletStorage
from electrum.util import MyEncoder
import json
import sys
import getpass
import os
default_fees= 100
def fix_will_settings_tx_fees(json_wallet):
tx_fees = json_wallet.get('will_settings',{}).get('tx_fees',False)
have_to_update=False
if tx_fees:
json_wallet['will_settings']['baltx_fees']=tx_fees
del json_wallet['will_settings']['tx_fees']
have_to_update=True
for txid,willitem in json_wallet['will'].items():
tx_fees=willitem.get('tx_fees',False)
if tx_fees:
json_wallet['will'][txid]['baltx_fees']=tx_fees
del json_wallet['will'][txid]['tx_fees']
have_to_update=True
return have_to_update
def uninstall_bal(json_wallet):
del json_wallet['will_settings']
del json_wallet['will']
del json_wallet['heirs']
return True
def save(json_wallet,storage):
human_readable=not storage.is_encrypted()
storage.write(json.dumps(
json_wallet,
indent=4 if human_readable else None,
sort_keys=bool(human_readable),
cls=MyEncoder,
))
def read_wallet(path,password=False):
storage=WalletStorage(path)
if storage.is_encrypted():
if password==False:
password = getpass.getpass("Enter wallet password: ", stream = None)
storage.decrypt(password)
data=storage.read()
json_wallet=json.loads('['+data+']')[0]
return json_wallet
if __name__ == '__main__':
if len(sys.argv) <3:
print("usage: ./bal_wallet_utils <command> <wallet path>")
print("available commands: uninstall, fix")
exit(1)
if not os.path.exists(sys.argv[2]):
print("Error: wallet not found")
exit(1)
command = sys.argv[1]
path = sys.argv[2]
json_wallet = read_wallet(path)
have_to_save=False
if command == 'fix':
have_to_save = fix_will_settings_tx_fees(json_wallet)
if command == 'uninstall':
have_to_save = uninstall_bal(json_wallet)
if have_to_save:
save(json_wallet,storage)
else:
print("nothing to do")

188
wallet_util/bal_wallet_utils_qt.py Executable file
View File

@ -0,0 +1,188 @@
#!/usr/bin/env python3
import sys
import os
import json
from PyQt6.QtWidgets import (QApplication, QMainWindow, QVBoxLayout, QHBoxLayout,
QLabel, QLineEdit, QPushButton, QWidget, QFileDialog,
QGroupBox, QTextEdit)
from PyQt6.QtCore import Qt
from electrum.storage import WalletStorage
from electrum.util import MyEncoder
from bal_wallet_utils import fix_will_settings_tx_fees,uninstall_bal,read_wallet
class WalletUtilityGUI(QMainWindow):
def __init__(self):
super().__init__()
self.initUI()
def initUI(self):
self.setWindowTitle('BAL Wallet Utility')
self.setFixedSize(500, 400)
# Central widget
central_widget = QWidget()
self.setCentralWidget(central_widget)
# Main layout
layout = QVBoxLayout(central_widget)
# Wallet input group
wallet_group = QGroupBox("Wallet Settings")
wallet_layout = QVBoxLayout(wallet_group)
# Wallet path
wallet_path_layout = QHBoxLayout()
wallet_path_layout.addWidget(QLabel("Wallet Path:"))
self.wallet_path_edit = QLineEdit()
self.wallet_path_edit.setPlaceholderText("Select wallet path...")
wallet_path_layout.addWidget(self.wallet_path_edit)
self.browse_btn = QPushButton("Browse...")
self.browse_btn.clicked.connect(self.browse_wallet)
wallet_path_layout.addWidget(self.browse_btn)
wallet_layout.addLayout(wallet_path_layout)
# Password
password_layout = QHBoxLayout()
password_layout.addWidget(QLabel("Password:"))
self.password_edit = QLineEdit()
self.password_edit.setEchoMode(QLineEdit.EchoMode.Password)
self.password_edit.setPlaceholderText("Enter password (if encrypted)")
password_layout.addWidget(self.password_edit)
wallet_layout.addLayout(password_layout)
layout.addWidget(wallet_group)
# Output area
output_group = QGroupBox("Output")
output_layout = QVBoxLayout(output_group)
self.output_text = QTextEdit()
self.output_text.setReadOnly(True)
output_layout.addWidget(self.output_text)
layout.addWidget(output_group)
# Action buttons
buttons_layout = QHBoxLayout()
self.fix_btn = QPushButton("Fix")
self.fix_btn.clicked.connect(self.fix_wallet)
self.fix_btn.setEnabled(False)
buttons_layout.addWidget(self.fix_btn)
self.uninstall_btn = QPushButton("Uninstall")
self.uninstall_btn.clicked.connect(self.uninstall_wallet)
self.uninstall_btn.setEnabled(False)
buttons_layout.addWidget(self.uninstall_btn)
layout.addLayout(buttons_layout)
# Connections to enable buttons when path is entered
self.wallet_path_edit.textChanged.connect(self.check_inputs)
def browse_wallet(self):
file_path, _ = QFileDialog.getOpenFileName(
self,
"Select Wallet",
"*",
"Electrum Wallet (*)"
)
if file_path:
self.wallet_path_edit.setText(file_path)
def check_inputs(self):
wallet_path = self.wallet_path_edit.text().strip()
has_path = bool(wallet_path) and os.path.exists(wallet_path)
self.fix_btn.setEnabled(has_path)
self.uninstall_btn.setEnabled(has_path)
def log_message(self, message):
self.output_text.append(message)
def fix_wallet(self):
self.process_wallet('fix')
def uninstall_wallet(self):
self.log_message("WARNING: This will remove all BAL settings. This operation cannot be undone.")
self.process_wallet('uninstall')
def process_wallet(self, command):
wallet_path = self.wallet_path_edit.text().strip()
password = self.password_edit.text()
if not wallet_path:
self.log_message("ERROR: Please enter wallet path")
return
if not os.path.exists(wallet_path):
self.log_message("ERROR: Wallet not found")
return
try:
self.log_message(f"Processing wallet: {wallet_path}")
storage = WalletStorage(wallet_path)
# Decrypt if necessary
if storage.is_encrypted():
if not password:
self.log_message("ERROR: Wallet is encrypted, please enter password")
return
try:
storage.decrypt(password)
self.log_message("Wallet decrypted successfully")
except Exception as e:
self.log_message(f"ERROR: Wrong password: {str(e)}")
return
# Read wallet
data = storage.read()
json_wallet = json.loads('[' + data + ']')[0]
have_to_save = False
message = ""
if command == 'fix':
have_to_save = fix_will_settings_tx_fees(json_wallet)
message = "Fix applied successfully" if have_to_save else "No fix needed"
elif command == 'uninstall':
have_to_save = uninstall_bal(json_wallet)
message = "BAL uninstalled successfully" if have_to_save else "No BAL settings found to uninstall"
if have_to_save:
try:
save_wallet(json_wallet, storage)
self.log_message(f"SUCCESS: {message}")
except Exception as e:
self.log_message(f"Save error: {str(e)}")
else:
self.log_message(f"INFO: {message}")
except Exception as e:
error_msg = f"ERROR: Processing failed: {str(e)}"
self.log_message(error_msg)
def main():
app = QApplication(sys.argv)
# Check if dependencies are available
try:
from electrum.storage import WalletStorage
from electrum.util import MyEncoder
except ImportError as e:
print(f"ERROR: Cannot import Electrum dependencies: {str(e)}")
return 1
window = WalletUtilityGUI()
window.show()
return app.exec()
if __name__ == '__main__':
sys.exit(main())

1064
will.py
View File

File diff suppressed because it is too large Load Diff

393
willexecutors.py
View File

@ -8,196 +8,249 @@ from electrum import constants
from electrum.logging import get_logger from electrum.logging import get_logger
from electrum.gui.qt.util import WaitingDialog from electrum.gui.qt.util import WaitingDialog
from electrum.i18n import _ from electrum.i18n import _
from .bal import BalPlugin
from .balqt.baldialog import BalWaitingDialog from .util import Util
from . import util as Util
DEFAULT_TIMEOUT = 5 DEFAULT_TIMEOUT = 5
_logger = get_logger(__name__) _logger = get_logger(__name__)
class Willexecutors:
def get_willexecutors(bal_plugin, update = False,bal_window=False,force=False,task=True): def save(bal_plugin, willexecutors):
willexecutors = bal_plugin.config_get(bal_plugin.WILLEXECUTORS) aw=bal_plugin.WILLEXECUTORS.get()
for w in willexecutors: aw[constants.net.NET_NAME]=willexecutors
initialize_willexecutor(willexecutors[w],w) bal_plugin.WILLEXECUTORS.set(aw)
bal=bal_plugin.DEFAULT_SETTINGS[bal_plugin.WILLEXECUTORS] def get_willexecutors(bal_plugin, update = False,bal_window=False,force=False,task=True):
for bal_url,bal_executor in bal.items(): willexecutors = bal_plugin.WILLEXECUTORS.get()
if not bal_url in willexecutors: willexecutors=willexecutors.get(constants.net.NET_NAME,{})
_logger.debug("replace bal") to_del=[]
willexecutors[bal_url]=bal_executor for w in willexecutors:
if update: if not isinstance(willexecutors[w],dict):
found = False to_del.append(w)
for url,we in willexecutors.items(): continue
if is_selected(we): Willexecutors.initialize_willexecutor(willexecutors[w],w)
found = True for w in to_del:
if found or force: print("ERROR: WILLEXECUTOR TO DELETE:", w)
if bal_plugin.config_get(bal_plugin.PING_WILLEXECUTORS) or force: del willexecutors[w]
ping_willexecutors = True bal = bal_plugin.WILLEXECUTORS.default.get(constants.net.NET_NAME,{})
if bal_plugin.config_get(bal_plugin.ASK_PING_WILLEXECUTORS) and not force: for bal_url,bal_executor in bal.items():
ping_willexecutors = bal_window.window.question(_("Contact willexecutors servers to update payment informations?")) if not bal_url in willexecutors:
if ping_willexecutors: _logger.debug(f"force add {bal_url} willexecutor")
if task: willexecutors[bal_url] = bal_executor
bal_window.ping_willexecutors(willexecutors) if update:
else: found = False
bal_window.ping_willexecutors_task(willexecutors) for url,we in willexecutors.items():
return willexecutors if Willexecutors.is_selected(we):
found = True
if found or force:
if bal_plugin.PING_WILLEXECUTORS.get() or force:
ping_willexecutors = True
if bal_plugin.ASK_PING_WILLEXECUTORS.get() and not force:
if bal_window:
ping_willexecutors = bal_window.window.question(_("Contact willexecutors servers to update payment informations?"))
def is_selected(willexecutor,value=None): if ping_willexecutors:
if not willexecutor: if task:
return False bal_window.ping_willexecutors(willexecutors,task)
if not value is None: else:
willexecutor['selected']=value bal_window.ping_willexecutors_task(willexecutors)
try: w_sorted = dict(sorted(willexecutors.items(), key=lambda w:w[1].get('sort',0),reverse=True))
return willexecutor['selected'] return w_sorted
except: def is_selected(willexecutor,value=None):
willexecutor['selected']=False if not willexecutor:
return False return False
if not value is None:
willexecutor['selected']=value
try:
return willexecutor['selected']
except:
willexecutor['selected']=False
return False
def get_willexecutor_transactions(will, force=False): def get_willexecutor_transactions(will, force=False):
willexecutors ={} willexecutors ={}
for wid,willitem in will.items(): for wid,willitem in will.items():
if willitem.get_status('VALID'): if willitem.get_status('VALID'):
if willitem.get_status('COMPLETE'): if willitem.get_status('COMPLETE'):
if not willitem.get_status('PUSHED') or force: if not willitem.get_status('PUSHED') or force:
if willexecutor := willitem.we: if willexecutor := willitem.we:
url=willexecutor['url'] url=willexecutor['url']
if willexecutor and is_selected(willexecutor): if willexecutor and Willexecutors.is_selected(willexecutor):
if not url in willexecutors: if not url in willexecutors:
willexecutor['txs']="" willexecutor['txs']=""
willexecutor['txsids']=[] willexecutor['txsids']=[]
willexecutor['broadcast_status']= _("Waiting...") willexecutor['broadcast_status']= _("Waiting...")
willexecutors[url]=willexecutor willexecutors[url]=willexecutor
willexecutors[url]['txs']+=str(willitem.tx)+"\n" willexecutors[url]['txs']+=str(willitem.tx)+"\n"
willexecutors[url]['txsids'].append(wid) willexecutors[url]['txsids'].append(wid)
return willexecutors return willexecutors
def only_selected_list(willexecutors): def only_selected_list(willexecutors):
out = {} out = {}
for url,v in willexectors.items(): for url,v in willexecutors.items():
if is_selected(willexecutor): if Willexecutors.is_selected(willexecutor):
out[url]=v out[url]=v
def push_transactions_to_willexecutors(will): def push_transactions_to_willexecutors(will):
willexecutors = get_transactions_to_be_pushed() willexecutors = get_transactions_to_be_pushed()
for url in willexecutors: for url in willexecutors:
willexecutor = willexecutors[url] willexecutor = willexecutors[url]
if is_selected(willexecutor): if Willexecutors.is_selected(willexecutor):
if 'txs' in willexecutor: if 'txs' in willexecutor:
push_transactions_to_willexecutor(willexecutors[url]['txs'],url) Willexecutors.push_transactions_to_willexecutor(willexecutors[url]['txs'],url)
def send_request(method, url, data=None, *, timeout=10): def send_request(method, url, data=None, *, timeout=10):
network = Network.get_instance() network = Network.get_instance()
if not network: if not network:
raise ErrorConnectingServer('You are offline.') raise ErrorConnectingServer('You are offline.')
_logger.debug(f'<-- {method} {url} {data}') _logger.debug(f'<-- {method} {url} {data}')
headers = {} headers = {}
headers['user-agent'] = 'BalPlugin' headers['user-agent'] = f"BalPlugin v:{BalPlugin.version()}"
headers['Content-Type']='text/plain' headers['Content-Type']='text/plain'
try: try:
if method == 'get': if method == 'get':
response = Network.send_http_on_proxy(method, url, response = Network.send_http_on_proxy(method, url,
params=data, params=data,
headers=headers, headers=headers,
on_finish=handle_response, on_finish=Willexecutors.handle_response,
timeout=timeout) timeout=timeout)
elif method == 'post': elif method == 'post':
response = Network.send_http_on_proxy(method, url, response = Network.send_http_on_proxy(method, url,
body=data, body=data,
headers=headers, headers=headers,
on_finish=handle_response, on_finish=Willexecutors.handle_response,
timeout=timeout) timeout=timeout)
else:
raise Exception(f"unexpected {method=!r}")
except Exception as e:
_logger.error(f"exception sending request {e}")
raise e
else: else:
raise Exception(f"unexpected {method=!r}") _logger.debug(f'--> {response}')
except Exception as e: return response
_logger.error(f"exception sending request {e}") async def handle_response(resp:ClientResponse):
raise e r=await resp.text()
else: try:
_logger.debug(f'--> {response}') r=json.loads(r)
return response r['status'] = resp.status
async def handle_response(resp:ClientResponse): r['selected']=Willexecutors.is_selected(willexecutor)
r=await resp.text() r['url']=url
try: except:
r=json.loads(r) pass
r['status'] = resp.status return r
r['selected']=is_selected(willexecutor)
r['url']=url class AlreadyPresentException(Exception):
except:
pass pass
return r def push_transactions_to_willexecutor(willexecutor):
out=True
try:
class AlreadyPresentException(Exception): _logger.debug(f"willexecutor['txs']")
pass if w:=Willexecutors.send_request('post', willexecutor['url']+"/"+constants.net.NET_NAME+"/pushtxs", data=willexecutor['txs'].encode('ascii')):
def push_transactions_to_willexecutor(willexecutor): willexecutor['broadcast_status'] = _("Success")
out=True _logger.debug(f"pushed: {w}")
try: if w !='thx':
_logger.debug(f"willexecutor['txs']") _logger.debug(f"error: {w}")
if w:=send_request('post', willexecutor['url']+"/"+constants.net.NET_NAME+"/pushtxs", data=willexecutor['txs'].encode('ascii')): raise Exception(w)
willexecutor['broadcast_stauts'] = _("Success") else:
_logger.debug(f"pushed: {w}") raise Exception("empty reply from:{willexecutor['url']}")
if w !='thx': except Exception as e:
_logger.debug(f"error: {w}") _logger.debug(f"error:{e}")
raise Exception(w) if str(e) == "already present":
else: raise Willexecutors.AlreadyPresentException()
raise Exception("empty reply from:{willexecutor['url']}") out=False
except Exception as e: willexecutor['broadcast_status'] = _("Failed")
_logger.debug(f"error:{e}")
if str(e) == "already present":
raise AlreadyPresentException()
out=False
willexecutor['broadcast_stauts'] = _("Failed")
return out return out
def ping_servers(willexecutors): def ping_servers(willexecutors):
for url,we in willexecutors.items(): for url,we in willexecutors.items():
get_info_task(url,we) Willexecutors.get_info_task(url,we)
def get_info_task(url,willexecutor): def get_info_task(url,willexecutor):
w=None w=None
try: try:
_logger.info("GETINFO_WILLEXECUTOR") _logger.info("GETINFO_WILLEXECUTOR")
_logger.debug(url) _logger.debug(url)
w = send_request('get',url+"/"+constants.net.NET_NAME+"/info") netname="bitcoin"
if constants.net.NET_NAME!="mainnet":
netname=constants.net.NET_NAME
w = Willexecutors.send_request('get',url+"/"+netname+"/info")
willexecutor['url']=url
willexecutor['status'] = w['status']
willexecutor['base_fee'] = w['base_fee']
willexecutor['address'] = w['address']
if not willexecutor['info']:
willexecutor['info'] = w['info']
_logger.debug(f"response_data {w['address']}")
except Exception as e:
_logger.error(f"error {e} contacting {url}: {w}")
willexecutor['status']="KO"
willexecutor['last_update'] = datetime.now().timestamp()
return willexecutor
def initialize_willexecutor(willexecutor,url,status=None,selected=None):
willexecutor['url']=url willexecutor['url']=url
willexecutor['status'] = w['status'] if not status is None:
willexecutor['base_fee'] = w['base_fee'] willexecutor['status'] = status
willexecutor['address'] = w['address'] willexecutor['selected'] = Willexecutors.is_selected(willexecutor,selected)
if not willexecutor['info']:
willexecutor['info'] = w['info']
_logger.debug(f"response_data {w['address']}")
except Exception as e:
_logger.error(f"error {e} contacting {url}: {w}")
willexecutor['status']="KO"
willexecutor['last_update'] = datetime.now().timestamp() def download_list(bal_plugin):
return willexecutor try:
l = Willexecutors.send_request('get',"https://welist.bitcoin-after.life/data/bitcoin?page=0&limit=100")
del l['status']
for w in l:
willexecutor=l[w]
Willexecutors.initialize_willexecutor(willexecutor,w,'New',False)
#bal_plugin.WILLEXECUTORS.set(l)
#bal_plugin.config.set_key(bal_plugin.WILLEXECUTORS,l,save=True)
return l
def initialize_willexecutor(willexecutor,url,status=None,selected=None): except Exception as e:
willexecutor['url']=url _logger.error(f"Failed to download willexecutors list: {e}")
if not status is None: return {}
willexecutor['status'] = status def get_willexecutors_list_from_json(bal_plugin):
willexecutor['selected'] = is_selected(willexecutor,selected) try:
with open("willexecutors.json") as f:
willexecutors = json.load(f)
for w in willexecutors:
willexecutor=willexecutors[w]
Willexecutors.initialize_willexecutor(willexecutor,w,'New',False)
#bal_plugin.WILLEXECUTORS.set(willexecutors)
return h
except Exception as e:
_logger.error(f"error opening willexecutors json: {e}")
return {}
def check_transaction(txid,url):
_logger.debug(f"{url}:{txid}")
try:
w = Willexecutors.send_request('post',url+"/searchtx",data=txid.encode('ascii'))
return w
except Exception as e:
_logger.error(f"error contacting {url} for checking txs {e}")
raise e
class WillExecutor:
def __init__(self,url,base_fee,chain,info,version):
self.url = url
self.base_fee = base_fee
self.chain = chain
self.info = info
self.version = version
def from_dict(d):
we = WillExecutor(
d['url'],
d['base_fee'],
d['chain'],
d['info'],
d['version']
)
def get_willexecutors_list_from_json(bal_plugin):
try:
with open("willexecutors.json") as f:
willexecutors = json.load(f)
for w in willexecutors:
willexecutor=willexecutors[w]
willexecutors.initialize_willexecutor(willexecutor,w,'New',False)
bal_plugin.config.set_key(bal_plugin.WILLEXECUTORS,willexecutors,save=True)
return h
except Exception as e:
_logger.error(f"errore aprendo willexecutors.json: {e}")
return {}
def check_transaction(txid,url):
_logger.debug(f"{url}:{txid}")
try:
w = send_request('post',url+"/searchtx",data=txid.encode('ascii'))
return w
except Exception as e:
_logger.error(f"error contacting {url} for checking txs {e}")
raise e