mmgen/mmgen-wallet
1
Fork 0
Code Releases 28 Wiki Activity

whitespace: tx

Browse source
This commit is contained in:
The MMGen Project 2024-10-18 10:32:10 +00:00
commit 362e581798
Signed by: mmgen
GPG key ID: 3F8B1861E32B7DA2
9 changed files with 135 additions and 135 deletions
Download patch file Download diff file Expand all files Collapse all files

22
mmgen/tx/__init__.py
View file

@ -14,14 +14,14 @@ tx.__init__: transaction class initializer
from ..objmethods import MMGenObject
def _base_proto_subclass(clsname,modname,proto):
def _base_proto_subclass(clsname, modname, proto):
if proto:
clsname = ('Token' if proto.tokensym else '') + clsname
modname = f'mmgen.proto.{proto.base_proto_coin.lower()}.tx.{modname}'
else:
modname = 'mmgen.tx.base'
import importlib
return getattr( importlib.import_module(modname), clsname )
return getattr(importlib.import_module(modname), clsname)
def _get_cls_info(clsname, modname, kwargs):
@ -31,22 +31,22 @@ def _get_cls_info(clsname, modname, kwargs):
proto = kwargs['data']['proto']
elif 'filename' in kwargs:
from .file import MMGenTxFile
proto = MMGenTxFile.get_proto( kwargs['cfg'], kwargs['filename'], quiet_open=True )
proto = MMGenTxFile.get_proto(kwargs['cfg'], kwargs['filename'], quiet_open=True)
elif clsname == 'Base':
proto = None
else:
raise ValueError(
f"{clsname} must be instantiated with 'proto','data' or 'filename' keyword" )
f"{clsname} must be instantiated with 'proto', 'data' or 'filename' keyword")
if clsname == 'Completed':
from ..util import get_extension,die
from ..util import get_extension, die
from .completed import Completed
ext = get_extension( kwargs['filename'] )
cls = Completed.ext_to_cls( ext, proto )
ext = get_extension(kwargs['filename'])
cls = Completed.ext_to_cls(ext, proto)
if not cls:
die(1,f'{ext!r}: unrecognized file extension for CompletedTX')
die(1, f'{ext!r}: unrecognized file extension for CompletedTX')
clsname = cls.__name__
modname = cls.__module__.rsplit('.',maxsplit=1)[-1]
modname = cls.__module__.rsplit('.', maxsplit=1)[-1]
kwargs['proto'] = proto
@ -86,10 +86,10 @@ async def _get_obj_async(_clsname, _modname, **kwargs):
return _base_proto_subclass(clsname, modname, proto)(**kwargs)
def _get(clsname,modname):
def _get(clsname, modname):
return lambda **kwargs: _get_obj(clsname, modname, **kwargs)
def _get_async(clsname,modname):
def _get_async(clsname, modname):
return lambda **kwargs: _get_obj_async(clsname, modname, **kwargs)
BaseTX = _get('Base', 'base')

76
mmgen/tx/base.py
View file

@ -25,24 +25,24 @@ from ..obj import (
NonNegativeInt
)
from ..amt import CoinAmtChk
from ..addr import MMGenID,CoinAddr
from ..util import msg,ymsg,fmt,remove_dups,make_timestamp,die
from ..addr import MMGenID, CoinAddr
from ..util import msg, ymsg, fmt, remove_dups, make_timestamp, die
class MMGenTxIO(MMGenListItem):
vout = ListItemAttr(NonNegativeInt)
amt = ImmutableAttr(CoinAmtChk, include_proto=True)
comment = ListItemAttr(TwComment,reassign_ok=True)
mmid = ListItemAttr(MMGenID,include_proto=True)
addr = ImmutableAttr(CoinAddr,include_proto=True)
comment = ListItemAttr(TwComment, reassign_ok=True)
mmid = ListItemAttr(MMGenID, include_proto=True)
addr = ImmutableAttr(CoinAddr, include_proto=True)
confs = ListItemAttr(int) # confs of type long exist in the wild, so convert
txid = ListItemAttr(CoinTxID)
have_wif = ListItemAttr(bool,typeconv=False,delete_ok=True)
have_wif = ListItemAttr(bool, typeconv=False, delete_ok=True)
invalid_attrs = {'proto','tw_copy_attrs'}
invalid_attrs = {'proto', 'tw_copy_attrs'}
def __init__(self,proto,**kwargs):
def __init__(self, proto, **kwargs):
self.__dict__['proto'] = proto
MMGenListItem.__init__(self,**kwargs)
MMGenListItem.__init__(self, **kwargs)
@property
def mmtype(self):
@ -55,17 +55,17 @@ class MMGenTxIO(MMGenListItem):
str(self.mmid.mmtype) if self.mmid else
'B' if self.addr.addr_fmt == 'bech32' else
'S' if self.addr.addr_fmt == 'p2sh' else
None )
None)
class MMGenTxIOList(list,MMGenObject):
class MMGenTxIOList(list, MMGenObject):
def __init__(self,parent,data=None):
def __init__(self, parent, data=None):
self.parent = parent
if data:
assert isinstance(data,list), 'MMGenTxIOList_check1'
assert isinstance(data, list), 'MMGenTxIOList_check1'
else:
data = []
list.__init__(self,data)
list.__init__(self, data)
class Base(MMGenObject):
desc = 'transaction'
@ -92,11 +92,11 @@ class Base(MMGenObject):
class Input(MMGenTxIO):
scriptPubKey = ListItemAttr(HexStr)
sequence = ListItemAttr(int,typeconv=False)
tw_copy_attrs = { 'scriptPubKey','vout','amt','comment','mmid','addr','confs','txid' }
sequence = ListItemAttr(int, typeconv=False)
tw_copy_attrs = {'scriptPubKey', 'vout', 'amt', 'comment', 'mmid', 'addr', 'confs', 'txid'}
class Output(MMGenTxIO):
is_chg = ListItemAttr(bool,typeconv=False)
is_chg = ListItemAttr(bool, typeconv=False)
class InputList(MMGenTxIOList):
desc = 'transaction inputs'
@ -104,7 +104,7 @@ class Base(MMGenObject):
class OutputList(MMGenTxIOList):
desc = 'transaction outputs'
def __init__(self,*args,**kwargs):
def __init__(self, *args, **kwargs):
self.cfg = kwargs['cfg']
self.inputs = self.InputList(self)
self.outputs = self.OutputList(self)
@ -126,15 +126,15 @@ class Base(MMGenObject):
return init_info(self.cfg, self)
def check_correct_chain(self):
if hasattr(self,'rpc'):
if hasattr(self, 'rpc'):
if self.chain != self.rpc.chain:
die( 'TransactionChainMismatch',
f'Transaction is for {self.chain}, but coin daemon chain is {self.rpc.chain}!' )
die('TransactionChainMismatch',
f'Transaction is for {self.chain}, but coin daemon chain is {self.rpc.chain}!')
def sum_inputs(self):
return sum(e.amt for e in self.inputs)
def sum_outputs(self,exclude=None):
def sum_outputs(self, exclude=None):
if exclude is None:
olist = self.outputs
else:
@ -143,12 +143,12 @@ class Base(MMGenObject):
return self.proto.coin_amt('0')
return sum(e.amt for e in olist)
def _chg_output_ops(self,op):
def _chg_output_ops(self, op):
is_chgs = [x.is_chg for x in self.outputs]
if is_chgs.count(True) == 1:
return (
is_chgs.index(True) if op == 'idx' else
self.outputs[is_chgs.index(True)] )
self.outputs[is_chgs.index(True)])
elif is_chgs.count(True) == 0:
return None
else:
@ -172,17 +172,17 @@ class Base(MMGenObject):
self.blockcount = self.rpc.blockcount
# returns True if comment added or changed, False otherwise
def add_comment(self,infile=None):
def add_comment(self, infile=None):
if infile:
from ..fileutil import get_data_from_file
self.comment = MMGenTxComment(get_data_from_file( self.cfg, infile, 'transaction comment' ))
self.comment = MMGenTxComment(get_data_from_file(self.cfg, infile, 'transaction comment'))
else:
from ..ui import keypress_confirm,line_input
from ..ui import keypress_confirm, line_input
if keypress_confirm(
self.cfg,
prompt = 'Edit transaction comment?' if self.comment else 'Add a comment to transaction?',
default_yes = False ):
res = MMGenTxComment(line_input( self.cfg, 'Comment: ', insert_txt=self.comment ))
default_yes = False):
res = MMGenTxComment(line_input(self.cfg, 'Comment: ', insert_txt=self.comment))
if not res:
ymsg('Warning: comment is empty')
changed = res != self.comment
@ -191,24 +191,24 @@ class Base(MMGenObject):
else:
return False
def get_non_mmaddrs(self,desc):
def get_non_mmaddrs(self, desc):
return remove_dups(
(i.addr for i in getattr(self,desc) if not i.mmid),
(i.addr for i in getattr(self, desc) if not i.mmid),
edesc = 'non-MMGen address',
quiet = True )
quiet = True)
def check_non_mmgen_inputs(self,caller,non_mmaddrs=None):
def check_non_mmgen_inputs(self, caller, non_mmaddrs=None):
non_mmaddrs = non_mmaddrs or self.get_non_mmaddrs('inputs')
if non_mmaddrs:
indent = ' '
fs = fmt(self.non_mmgen_inputs_msg,strip_char='\t',indent=indent).strip()
fs = fmt(self.non_mmgen_inputs_msg, strip_char='\t', indent=indent).strip()
m = fs.format('\n '.join(non_mmaddrs))
if caller in ('txdo','txsign'):
if caller in ('txdo', 'txsign'):
if not self.cfg.keys_from_file:
die( 'UserOptError', f'\n{indent}ERROR: {m}\n' )
die('UserOptError', f'\n{indent}ERROR: {m}\n')
else:
msg(f'\n{indent}WARNING: {m}\n')
if not self.cfg.yes:
from ..ui import keypress_confirm
if not keypress_confirm( self.cfg, 'Continue?', default_yes=True ):
die(1,'Exiting at user request')
if not keypress_confirm(self.cfg, 'Continue?', default_yes=True):
die(1, 'Exiting at user request')

20
mmgen/tx/bump.py
View file

@ -14,23 +14,23 @@ tx.bump: transaction bump class
from .new import New
from .completed import Completed
from ..util import msg,is_int,die
from ..util import msg, is_int, die
class Bump(Completed,New):
class Bump(Completed, New):
desc = 'fee-bumped transaction'
ext = 'rawtx'
bump_output_idx = None
def __init__(self,check_sent,*args,**kwargs):
def __init__(self, check_sent, *args, **kwargs):
super().__init__(*args,**kwargs)
super().__init__(*args, **kwargs)
if not self.is_replaceable():
die(1,f'Transaction {self.txid} is not replaceable')
die(1, f'Transaction {self.txid} is not replaceable')
# If sending, require original tx to be sent
if check_sent and not self.coin_txid:
die(1,'Transaction {self.txid!r} was not broadcast to the network')
die(1, 'Transaction {self.txid!r} was not broadcast to the network')
self.coin_txid = ''
self.sent_timestamp = None
@ -54,7 +54,7 @@ class Bump(Completed,New):
self.bump_output_idx = 0
return 0
else:
die(1,'Insufficient funds to bump transaction')
die(1, 'Insufficient funds to bump transaction')
init_reply = self.cfg.output_to_reduce
chg_idx = self.chg_idx
@ -63,9 +63,9 @@ class Bump(Completed,New):
if init_reply is None:
from ..ui import line_input
m = 'Choose an output to deduct the fee from (Hit ENTER for the change output): '
reply = line_input( self.cfg, m ) or 'c'
reply = line_input(self.cfg, m) or 'c'
else:
reply,init_reply = init_reply,None
reply, init_reply = init_reply, None
if chg_idx is None and not is_int(reply):
msg('Output must be an integer')
elif chg_idx is not None and not is_int(reply) and reply != 'c':
@ -83,7 +83,7 @@ class Bump(Completed,New):
msg(prompt)
else:
from ..ui import keypress_confirm
if not keypress_confirm( self.cfg, prompt+'. OK?', default_yes=True ):
if not keypress_confirm(self.cfg, prompt+'. OK?', default_yes=True):
continue
self.bump_output_idx = idx
return idx

6
mmgen/tx/completed.py
View file

@ -20,11 +20,11 @@ class Completed(Base):
"""
filename_api = True
def __init__(self,cfg,*args,filename=None,data=None,quiet_open=False,**kwargs):
def __init__(self, cfg, *args, filename=None, data=None, quiet_open=False, **kwargs):
assert (filename or data) and not (filename and data), 'CompletedTX_chk1'
super().__init__(cfg=cfg,*args,**kwargs)
super().__init__(cfg=cfg, *args, **kwargs)
if data:
self.__dict__ = data | {'twctl': self.twctl}
@ -40,7 +40,7 @@ class Completed(Base):
if self.check_sigs() != self.signed:
from ..util import die
die(1,'Transaction is {}signed!'.format('not ' if self.signed else ''))
die(1, 'Transaction is {}signed!'.format('not ' if self.signed else ''))
self.infile = filename

42
mmgen/tx/info.py
View file

@ -15,8 +15,8 @@ tx.info: transaction info class
import importlib
from ..cfg import gc
from ..color import red,green,orange
from ..util import msg,msg_r,decode_timestamp,make_timestr
from ..color import red, green, orange
from ..util import msg, msg_r, decode_timestamp, make_timestr
from ..util2 import format_elapsed_hr
class TxInfo:
@ -40,11 +40,11 @@ class TxInfo:
def get_max_mmwid(io):
sel_f = (
(lambda o: len(o.mmid) + 2) if io == tx.inputs else # 2 = len('()')
(lambda o: len(o.mmid) + (2,8)[bool(o.is_chg)]) ) # 6 = len(' (chg)')
return max(max([sel_f(o) for o in io if o.mmid] or [0]),len(nonmm_str))
(lambda o: len(o.mmid) + (2, 8)[bool(o.is_chg)])) # 6 = len(' (chg)')
return max(max([sel_f(o) for o in io if o.mmid] or [0]), len(nonmm_str))
nonmm_str = f'(non-{gc.proj_name} address)'
max_mmwid = max(get_max_mmwid(tx.inputs),get_max_mmwid(tx.outputs))
max_mmwid = max(get_max_mmwid(tx.inputs), get_max_mmwid(tx.outputs))
def gen_view():
yield (self.txinfo_hdr_fs_short if terse else self.txinfo_hdr_fs).format(
@ -55,10 +55,10 @@ class TxInfo:
s = green('True') if tx.signed else red('False'),
l = (
orange(self.strfmt_locktime(terse=True)) if tx.locktime else
green('None') ))
green('None')))
for attr,label in [('timestamp','Created:'),('sent_timestamp','Sent:')]:
if (val := getattr(tx,attr)) is not None:
for attr, label in [('timestamp', 'Created:'), ('sent_timestamp', 'Sent:')]:
if (val := getattr(tx, attr)) is not None:
_ = decode_timestamp(val)
yield f'{label:8} {make_timestr(_)} ({format_elapsed_hr(_)})\n'
@ -68,7 +68,7 @@ class TxInfo:
if tx.coin_txid:
yield f'{tx.coin} TxID: {tx.coin_txid.hl()}\n'
enl = ('\n','')[bool(terse)]
enl = ('\n', '')[bool(terse)]
yield enl
if tx.comment:
@ -85,25 +85,25 @@ class TxInfo:
iwidth = len(str(int(tx.sum_inputs())))
yield self.txinfo_ftr_fs.format(
i = tx.sum_inputs().fmt(color=True,iwidth=iwidth),
o = tx.sum_outputs().fmt(color=True,iwidth=iwidth),
C = tx.change.fmt(color=True,iwidth=iwidth),
s = tx.send_amt.fmt(color=True,iwidth=iwidth),
a = self.format_abs_fee(color=True,iwidth=iwidth),
i = tx.sum_inputs().fmt(color=True, iwidth=iwidth),
o = tx.sum_outputs().fmt(color=True, iwidth=iwidth),
C = tx.change.fmt(color=True, iwidth=iwidth),
s = tx.send_amt.fmt(color=True, iwidth=iwidth),
a = self.format_abs_fee(color=True, iwidth=iwidth),
r = self.format_rel_fee(),
d = tx.dcoin,
c = tx.coin )
c = tx.coin)
if tx.cfg.verbose:
yield self.format_verbose_footer()
return ''.join(gen_view()) # TX label might contain non-ascii chars
return ''.join(gen_view())
def view_with_prompt(self,prompt,pause=True):
def view_with_prompt(self, prompt, pause=True):
prompt += ' (y)es, (N)o, pager (v)iew, (t)erse view: '
from ..term import get_char
while True:
reply = get_char( prompt, immed_chars='YyNnVvTt' ).strip('\n\r')
reply = get_char(prompt, immed_chars='YyNnVvTt').strip('\n\r')
msg('')
if reply == '' or reply in 'Nn':
break
@ -111,11 +111,11 @@ class TxInfo:
self.view(
pager = reply in 'Vv',
pause = pause,
terse = reply in 'Tt' )
terse = reply in 'Tt')
break
msg('Invalid reply')
def view(self,pager=False,pause=True,terse=False):
def view(self, pager=False, pause=True, terse=False):
o = self.format(terse=terse)
if pager:
from ..ui import do_pager
@ -130,4 +130,4 @@ class TxInfo:
def init_info(cfg, tx):
return getattr(
importlib.import_module(f'mmgen.proto.{tx.proto.base_proto_coin.lower()}.tx.info'),
('Token' if tx.proto.tokensym else '') + 'TxInfo' )(cfg, tx)
('Token' if tx.proto.tokensym else '') + 'TxInfo')(cfg, tx)

4
mmgen/tx/online.py
View file

@ -19,7 +19,7 @@ class OnlineSigned(Signed):
@property
def status(self):
from . import _base_proto_subclass
return _base_proto_subclass('Status','status',self.proto)(self)
return _base_proto_subclass('Status', 'status', self.proto)(self)
def confirm_send(self):
from ..util import msg
@ -28,7 +28,7 @@ class OnlineSigned(Signed):
cfg = self.cfg,
message = '' if self.cfg.quiet else 'Once this transaction is sent, there’s no taking it back!',
action = f'broadcast this transaction to the {self.proto.coin} {self.proto.network.upper()} network',
expect = 'YES' if self.cfg.quiet or self.cfg.yes else 'YES, I REALLY WANT TO DO THIS' )
expect = 'YES' if self.cfg.quiet or self.cfg.yes else 'YES, I REALLY WANT TO DO THIS')
msg('Sending transaction')
class AutomountOnlineSigned(AutomountSigned, OnlineSigned):

84
mmgen/tx/sign.py
View file

@ -21,15 +21,15 @@ tx.sign: Sign a transaction generated by 'mmgen-txcreate'
"""
from ..cfg import gc
from ..util import msg,suf,fmt,die,remove_dups,get_extension
from ..util import msg, suf, fmt, die, remove_dups, get_extension
from ..obj import MMGenList
from ..addr import MMGenAddrType
from ..addrlist import AddrIdxList,KeyAddrList
from ..wallet import Wallet,get_wallet_extensions,get_wallet_cls
from ..addrlist import AddrIdxList, KeyAddrList
from ..wallet import Wallet, get_wallet_extensions, get_wallet_cls
saved_seeds = {}
def get_seed_for_seed_id(sid,infiles,saved_seeds):
def get_seed_for_seed_id(sid, infiles, saved_seeds):
if sid in saved_seeds:
return saved_seeds[sid]
@ -39,7 +39,7 @@ def get_seed_for_seed_id(sid,infiles,saved_seeds):
if infiles:
seed = Wallet(cfg, infiles.pop(0), ignore_in_fmt=True, passwd_file=global_passwd_file).seed
elif subseeds_checked is False:
seed = saved_seeds[list(saved_seeds)[0]].subseed_by_seed_id(sid,print_msg=True)
seed = saved_seeds[list(saved_seeds)[0]].subseed_by_seed_id(sid, print_msg=True)
subseeds_checked = True
if not seed:
continue
@ -48,23 +48,23 @@ def get_seed_for_seed_id(sid,infiles,saved_seeds):
seed = Wallet(cfg, passwd_file=global_passwd_file).seed
msg(f'User input produced Seed ID {seed.sid}')
if not seed.sid == sid: # TODO: add test
seed = seed.subseed_by_seed_id(sid,print_msg=True)
seed = seed.subseed_by_seed_id(sid, print_msg=True)
if seed:
saved_seeds[seed.sid] = seed
if seed.sid == sid:
return seed
else:
die(2,f'ERROR: No seed source found for Seed ID: {sid}')
die(2, f'ERROR: No seed source found for Seed ID: {sid}')
def generate_kals_for_mmgen_addrs(need_keys,infiles,saved_seeds,proto):
def generate_kals_for_mmgen_addrs(need_keys, infiles, saved_seeds, proto):
mmids = [e.mmid for e in need_keys]
sids = remove_dups((i.sid for i in mmids),quiet=True)
sids = remove_dups((i.sid for i in mmids), quiet=True)
cfg._util.vmsg(f"Need seed{suf(sids)}: {' '.join(sids)}")
def gen_kals():
for sid in sids:
# Returns only if seed is found
seed = get_seed_for_seed_id(sid,infiles,saved_seeds)
seed = get_seed_for_seed_id(sid, infiles, saved_seeds)
for id_str in MMGenAddrType.mmtypes:
idx_list = [i.idx for i in mmids if i.sid == sid and i.mmtype == id_str]
if idx_list:
@ -73,21 +73,21 @@ def generate_kals_for_mmgen_addrs(need_keys,infiles,saved_seeds,proto):
proto = proto,
seed = seed,
addr_idxs = AddrIdxList(idx_list=idx_list),
mmtype = MMGenAddrType(proto,id_str),
skip_chksum = True )
mmtype = MMGenAddrType(proto, id_str),
skip_chksum = True)
return MMGenList(gen_kals())
def add_keys(tx,src,infiles=None,saved_seeds=None,keyaddr_list=None):
need_keys = [e for e in getattr(tx,src) if e.mmid and not e.have_wif]
def add_keys(tx, src, infiles=None, saved_seeds=None, keyaddr_list=None):
need_keys = [e for e in getattr(tx, src) if e.mmid and not e.have_wif]
if not need_keys:
return []
desc,src_desc = (
('key-address file','From key-address file:') if keyaddr_list else
('seed(s)','Generated from seed:') )
desc, src_desc = (
('key-address file', 'From key-address file:') if keyaddr_list else
('seed(s)', 'Generated from seed:'))
cfg._util.qmsg(f'Checking {gc.proj_name} -> {tx.proto.coin} address mappings for {src} (from {desc})')
d = (
MMGenList([keyaddr_list]) if keyaddr_list else
generate_kals_for_mmgen_addrs(need_keys,infiles,saved_seeds,tx.proto) )
generate_kals_for_mmgen_addrs(need_keys, infiles, saved_seeds, tx.proto))
new_keys = []
for e in need_keys:
for kal in d:
@ -99,7 +99,7 @@ def add_keys(tx,src,infiles=None,saved_seeds=None,keyaddr_list=None):
if src == 'inputs':
new_keys.append(f)
else:
die(3,fmt(f"""
die(3, fmt(f"""
{gc.proj_name} -> {tx.proto.coin} address mappings differ!
{src_desc:<23} {mmid} -> {f.addr}
{'tx file:':<23} {e.mmid} -> {e.addr}
@ -108,37 +108,37 @@ def add_keys(tx,src,infiles=None,saved_seeds=None,keyaddr_list=None):
cfg._util.vmsg(f'Added {len(new_keys)} wif key{suf(new_keys)} from {desc}')
return new_keys
def _pop_matching_fns(args,cmplist): # strips found args
def _pop_matching_fns(args, cmplist): # strips found args
return list(reversed([args.pop(args.index(a)) for a in reversed(args) if get_extension(a) in cmplist]))
def get_tx_files(cfg, args):
from .unsigned import Unsigned, AutomountUnsigned
ret = _pop_matching_fns(args, [(AutomountUnsigned if cfg.autosign else Unsigned).ext])
if not ret:
die(1,'You must specify a raw transaction file!')
die(1, 'You must specify a raw transaction file!')
return ret
def get_seed_files(cfg,args):
def get_seed_files(cfg, args):
# favor unencrypted seed sources first, as they don't require passwords
ret = _pop_matching_fns( args, get_wallet_extensions('unenc') )
ret = _pop_matching_fns(args, get_wallet_extensions('unenc'))
from ..filename import find_file_in_dir
wf = find_file_in_dir(get_wallet_cls('mmgen'),cfg.data_dir) # Make this the first encrypted ss in the list
wf = find_file_in_dir(get_wallet_cls('mmgen'), cfg.data_dir) # Make this the first encrypted ss in the list
if wf:
ret.append(wf)
ret += _pop_matching_fns( args, get_wallet_extensions('enc') )
ret += _pop_matching_fns(args, get_wallet_extensions('enc'))
if not (ret or cfg.mmgen_keys_from_file or cfg.keys_from_file): # or cfg.use_wallet_dat
die(1,'You must specify a seed or key source!')
die(1, 'You must specify a seed or key source!')
return ret
def get_keyaddrlist(cfg,proto):
def get_keyaddrlist(cfg, proto):
if cfg.mmgen_keys_from_file:
return KeyAddrList( cfg, proto, cfg.mmgen_keys_from_file )
return KeyAddrList(cfg, proto, cfg.mmgen_keys_from_file)
return None
def get_keylist(cfg):
if cfg.keys_from_file:
from ..fileutil import get_lines_from_file
return get_lines_from_file( cfg, cfg.keys_from_file, 'key-address data', trim_comments=True )
return get_lines_from_file(cfg, cfg.keys_from_file, 'key-address data', trim_comments=True)
return None
async def txsign(cfg_parm, tx, seed_files, kl, kal, tx_num_str='', passwd_file=None):
@ -151,39 +151,39 @@ async def txsign(cfg_parm, tx, seed_files, kl, kal, tx_num_str='', passwd_file=N
global_passwd_file = passwd_file
if non_mmaddrs:
tx.check_non_mmgen_inputs(caller='txsign',non_mmaddrs=non_mmaddrs)
tx.check_non_mmgen_inputs(caller='txsign', non_mmaddrs=non_mmaddrs)
tmp = KeyAddrList(
cfg = cfg,
proto = tx.proto,
addrlist = non_mmaddrs,
skip_chksum = True )
skip_chksum = True)
if kl:
tmp.add_wifs(kl)
missing = tmp.list_missing('sec')
if missing:
sep = '\n '
die(2,'ERROR: a key file must be supplied for the following non-{} address{}:{}'.format(
die(2, 'ERROR: a key file must be supplied for the following non-{} address{}:{}'.format(
gc.proj_name,
suf(missing,'es'),
sep + sep.join(missing) ))
suf(missing, 'es'),
sep + sep.join(missing)))
keys += tmp.data
if cfg.mmgen_keys_from_file:
keys += add_keys(tx,'inputs',keyaddr_list=kal)
add_keys(tx,'outputs',keyaddr_list=kal)
keys += add_keys(tx, 'inputs', keyaddr_list=kal)
add_keys(tx, 'outputs', keyaddr_list=kal)
keys += add_keys(tx,'inputs',seed_files,saved_seeds)
add_keys(tx,'outputs',seed_files,saved_seeds)
keys += add_keys(tx, 'inputs', seed_files, saved_seeds)
add_keys(tx, 'outputs', seed_files, saved_seeds)
# this (boolean) attr isn't needed in transaction file
tx.delete_attrs('inputs','have_wif')
tx.delete_attrs('outputs','have_wif')
tx.delete_attrs('inputs', 'have_wif')
tx.delete_attrs('outputs', 'have_wif')
extra_sids = remove_dups(
(s for s in saved_seeds if s not in tx.get_sids('inputs') + tx.get_sids('outputs')),
quiet = True )
quiet = True)
if extra_sids:
msg(f"Unused Seed ID{suf(extra_sids)}: {' '.join(extra_sids)}")
return await tx.sign(tx_num_str,keys) # returns signed TX object or False
return await tx.sign(tx_num_str, keys) # returns signed TX object or False

2
mmgen/tx/status.py
View file

@ -14,5 +14,5 @@ tx.status: transaction status class
class Status:
def __init__(self,parent_tx):
def __init__(self, parent_tx):
self.tx = parent_tx

14
mmgen/tx/unsigned.py
View file

@ -20,15 +20,15 @@ class Unsigned(Completed):
ext = 'rawtx'
automount = False
def delete_attrs(self,desc,attr):
for e in getattr(self,desc):
if hasattr(e,attr):
delattr(e,attr)
def delete_attrs(self, desc, attr):
for e in getattr(self, desc):
if hasattr(e, attr):
delattr(e, attr)
def get_sids(self,desc):
def get_sids(self, desc):
return remove_dups(
(e.mmid.sid for e in getattr(self,desc) if e.mmid),
quiet = True )
(e.mmid.sid for e in getattr(self, desc) if e.mmid),
quiet = True)
class AutomountUnsigned(Unsigned):
desc = 'unsigned automount transaction'