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

f-strings, whitespace (program files) [43 files patched]

Browse source
This commit is contained in:
The MMGen Project 2021-09-29 21:17:56 +00:00
commit 2872d4b683
Signed by: mmgen
GPG key ID: 3F8B1861E32B7DA2
43 changed files with 614 additions and 525 deletions
Download patch file Download diff file Expand all files Collapse all files

118
mmgen/addr.py
View file

@ -347,7 +347,11 @@ class AddrListIDStr(str,Hilite):
else:
bc = (addrlist.proto.base_coin,addrlist.proto.coin)[addrlist.proto.base_coin=='ETH']
mt = addrlist.al_id.mmtype
ret = '{}{}{}[{}]'.format(addrlist.al_id.sid,('-'+bc,'')[bc=='BTC'],('-'+mt,'')[mt in ('L','E')],s)
ret = '{}{}{}[{}]'.format(
addrlist.al_id.sid,
('-'+bc,'')[bc == 'BTC'],
('-'+mt,'')[mt in ('L','E')],
s )
dmsg_sc('id_str',ret[8:].split('[')[0])
@ -355,7 +359,7 @@ class AddrListIDStr(str,Hilite):
class AddrList(MMGenObject): # Address info for a single seed ID
msgs = {
'file_header': """
'file_header': """
# {pnm} address file
#
# This file is editable.
@ -364,13 +368,13 @@ class AddrList(MMGenObject): # Address info for a single seed ID
# address, and it will be appended to the tracking wallet label upon import.
# The label may contain any printable ASCII symbol.
""".strip().format(n=TwComment.max_screen_width,pnm=pnm),
'record_chksum': """
'record_chksum': """
Record this checksum: it will be used to verify the address file in the future
""".strip(),
'check_chksum': 'Check this value against your records',
'removed_dup_keys': """
'check_chksum': 'Check this value against your records',
'removed_dup_keys': f"""
Removed {{}} duplicate WIF key{{}} from keylist (also in {pnm} key-address file
""".strip().format(pnm=pnm)
""".strip(),
}
entry_type = AddrListEntry
main_attr = 'addr'
@ -431,7 +435,7 @@ Removed {{}} duplicate WIF key{{}} from keylist (also in {pnm} key-address file
elif al_id or adata:
die(3,'Must specify both al_id and adata')
else:
die(3,'Incorrect arguments for {}'.format(type(self).__name__))
die(3,f'Incorrect arguments for {type(self).__name__}')
# al_id,adata now set
self.data = adata
@ -446,8 +450,7 @@ Removed {{}} duplicate WIF key{{}} from keylist (also in {pnm} key-address file
if do_chksum:
self.chksum = AddrListChksum(self)
qmsg('Checksum for {} data {}: {}'.format(
self.data_desc,self.id_str.hl(),self.chksum.hl()))
qmsg(f'Checksum for {self.data_desc} data {self.id_str.hl()}: {self.chksum.hl()}')
qmsg(self.msgs[('check_chksum','record_chksum')[src=='gen']])
def update_msgs(self):
@ -482,7 +485,7 @@ Removed {{}} duplicate WIF key{{}} from keylist (also in {pnm} key-address file
pos += 1
if not g.debug:
qmsg_r('\rGenerating {} #{} ({} of {})'.format(self.gen_desc,num,pos,t_addrs))
qmsg_r(f'\rGenerating {self.gen_desc} #{num} ({pos} of {t_addrs})')
e = le(proto=self.proto,idx=num)
@ -503,14 +506,19 @@ Removed {{}} duplicate WIF key{{}} from keylist (also in {pnm} key-address file
if type(self) == PasswordList:
e.passwd = str(self.make_passwd(e.sec)) # TODO - own type
dmsg('Key {:>03}: {}'.format(pos,e.passwd))
dmsg(f'Key {pos:>03}: {e.passwd}')
out.append(e)
if g.debug_addrlist:
Msg('generate():\n{}'.format(e.pfmt()))
Msg(f'generate():\n{e.pfmt()}')
qmsg('\r{}: {} {}{} generated{}'.format(
self.al_id.hl(),t_addrs,self.gen_desc,suf(t_addrs,self.gen_desc_pl),' '*15))
self.al_id.hl(),
t_addrs,
self.gen_desc,
suf(t_addrs,self.gen_desc_pl),
' ' * 15 ))
return out
def check_format(self,addr):
@ -546,7 +554,7 @@ Removed {{}} duplicate WIF key{{}} from keylist (also in {pnm} key-address file
return [e.idx for e in self.data]
def addrs(self):
return ['{}:{}'.format(self.al_id.sid,e.idx) for e in self.data]
return [f'{self.al_id.sid}:{e.idx}' for e in self.data]
def addrpairs(self):
return [(e.idx,e.addr) for e in self.data]
@ -612,11 +620,11 @@ Removed {{}} duplicate WIF key{{}} from keylist (also in {pnm} key-address file
ag = AddrGenerator(self.proto,at)
d = self.data
for n,e in enumerate(d,1):
qmsg_r('\rGenerating addresses from keylist: {}/{}'.format(n,len(d)))
qmsg_r(f'\rGenerating addresses from keylist: {n}/{len(d)}')
e.addr = ag.to_addr(kg.to_pubhex(e.sec))
if g.debug_addrlist:
Msg('generate_addrs_from_keys():\n{}'.format(e.pfmt()))
qmsg('\rGenerated addresses from keylist: {}/{} '.format(n,len(d)))
Msg(f'generate_addrs_from_keys():\n{e.pfmt()}')
qmsg(f'\rGenerated addresses from keylist: {n}/{len(d)} ')
def make_label(self):
bc,mt = self.proto.base_coin,self.al_id.mmtype
@ -630,30 +638,29 @@ Removed {{}} duplicate WIF key{{}} from keylist (also in {pnm} key-address file
out = [self.msgs['file_header']+'\n']
if self.chksum:
out.append('# {} data checksum for {}: {}'.format(
capfirst(self.data_desc),self.id_str,self.chksum))
out.append(f'# {capfirst(self.data_desc)} data checksum for {self.id_str}: {self.chksum}')
out.append('# Record this value to a secure location.\n')
lbl = self.make_label()
dmsg_sc('lbl',lbl[9:])
out.append('{} {{'.format(lbl))
out.append(f'{lbl} {{')
fs = ' {:<%s} {:<34}{}' % len(str(self.data[-1].idx))
for e in self.data:
c = ' '+e.label if add_comments and e.label else ''
if type(self) == KeyList:
out.append(fs.format(e.idx,'{}: {}'.format(self.al_id.mmtype.wif_label,e.sec.wif),c))
out.append(fs.format( e.idx, f'{self.al_id.mmtype.wif_label}: {e.sec.wif}', c ))
elif type(self) == PasswordList:
out.append(fs.format(e.idx,e.passwd,c))
else: # First line with idx
out.append(fs.format(e.idx,e.addr,c))
if self.has_keys:
if opt.b16:
out.append(fs.format('', 'orig_hex: '+e.sec.orig_hex,c))
out.append(fs.format('','{}: {}'.format(self.al_id.mmtype.wif_label,e.sec.wif),c))
out.append(fs.format( '', f'orig_hex: {e.sec.orig_hex}', c ))
out.append(fs.format( '', f'{self.al_id.mmtype.wif_label}: {e.sec.wif}', c ))
for k in ('viewkey','wallet_passwd'):
v = getattr(e,k)
if v: out.append(fs.format('','{}: {}'.format(k,v),c))
if v: out.append(fs.format( '', f'{k}: {v}', c ))
out.append('}')
self.fmt_data = '\n'.join([l.rstrip() for l in out]) + '\n'
@ -682,14 +689,14 @@ Removed {{}} duplicate WIF key{{}} from keylist (also in {pnm} key-address file
if self.has_keys: # order: wif,(orig_hex),viewkey,wallet_passwd
d = self.get_line(lines)
assert d[0] == self.al_id.mmtype.wif_label+':',iifs.format(d[0],self.al_id.mmtype.wif_label)
assert d[0] == self.al_id.mmtype.wif_label+':', iifs.format(d[0],self.al_id.mmtype.wif_label)
a.sec = PrivKey(proto=self.proto,wif=d[1])
for k,dtype,add_proto in (
('viewkey',ViewKey,True),
('wallet_passwd',WalletPassword,False) ):
if k in self.al_id.mmtype.extra_attrs:
d = self.get_line(lines)
assert d[0] == k+':',iifs.format(d[0],k)
assert d[0] == k+':', iifs.format(d[0],k)
setattr(a,k,dtype( *((self.proto,d[1]) if add_proto else (d[1],)) ) )
ret.append(a)
@ -700,9 +707,9 @@ Removed {{}} duplicate WIF key{{}} from keylist (also in {pnm} key-address file
ag = AddrGenerator(self.proto,self.al_id.mmtype)
llen = len(ret)
for n,e in enumerate(ret):
qmsg_r('\rVerifying keys {}/{}'.format(n+1,llen))
qmsg_r(f'\rVerifying keys {n+1}/{llen}')
assert e.addr == ag.to_addr(kg.to_pubhex(e.sec)),(
"Key doesn't match address!\n {}\n {}".format(e.sec.wif,e.addr))
f'Key doesn’t match address!\n {e.sec.wif}\n {e.addr}')
qmsg(' - done')
return ret
@ -934,9 +941,9 @@ Record this checksum: it will be used to verify the password file in the future
self.fmt_data = ''
self.chksum = AddrListChksum(self)
fs = '{}-{}-{}-{}[{{}}]'.format(self.al_id.sid,self.pw_id_str,self.pw_fmt_disp,self.pw_len)
fs = f'{self.al_id.sid}-{self.pw_id_str}-{self.pw_fmt_disp}-{self.pw_len}[{{}}]'
self.id_str = AddrListIDStr(self,fs)
qmsg('Checksum for {} data {}: {}'.format(self.data_desc,self.id_str.hl(),self.chksum.hl()))
qmsg(f'Checksum for {self.data_desc} data {self.id_str.hl()}: {self.chksum.hl()}')
qmsg(self.msgs[('record_chksum','check_chksum')[bool(infile)]])
def set_pw_fmt(self,pw_fmt):
@ -964,11 +971,11 @@ Record this checksum: it will be used to verify the password file in the future
d = self.pw_info[self.pw_fmt]
if d.valid_lens:
if pw_len not in d.valid_lens:
die(2,fs.format(l=pw_len,b=d.desc,c='not one of ',m=d.valid_lens,pw=passwd))
die(2, fs.format( l=pw_len, b=d.desc, c='not one of ', m=d.valid_lens, pw=passwd ))
elif pw_len > d.max_len:
die(2,fs.format(l=pw_len,b=d.desc,c='>',m=d.max_len,pw=passwd))
die(2, fs.format( l=pw_len, b=d.desc, c='>', m=d.max_len, pw=passwd ))
elif pw_len < d.min_len:
die(2,fs.format(l=pw_len,b=d.desc,c='<',m=d.min_len,pw=passwd))
die(2, fs.format( l=pw_len, b=d.desc, c='<', m=d.min_len, pw=passwd ))
def set_pw_len(self,pw_len):
d = self.pw_info[self.pw_fmt]
@ -978,7 +985,7 @@ Record this checksum: it will be used to verify the password file in the future
return
if not is_int(pw_len):
die(2,"'{}': invalid user-requested password length (not an integer)".format(pw_len,d.desc))
die(2,f'{pw_len!r}: invalid user-requested password length (not an integer)')
self.pw_len = int(pw_len)
self.chk_pw_len()
@ -996,24 +1003,27 @@ Record this checksum: it will be used to verify the password file in the future
try:
good_pw_len = baseconv.seedlen_map['xmrseed'][seed.byte_len]
except:
die(1,'{}: unsupported seed length for Monero new-style mnemonic'.format(seed.byte_len*8))
die(1,f'{seed.byte_len*8}: unsupported seed length for Monero new-style mnemonic')
elif pf in ('b32','b58'):
pw_int = (32 if pf == 'b32' else 58) ** self.pw_len
pw_bytes = pw_int.bit_length() // 8
good_pw_len = len(baseconv.frombytes(b'\xff'*seed.byte_len,wl_id=pf))
else:
raise NotImplementedError('{!r}: unknown password format'.format(pf))
raise NotImplementedError(f'{pf!r}: unknown password format')
if pw_bytes > seed.byte_len:
m1 = 'Cannot generate passwords with more entropy than underlying seed! ({} bits)'
m2 = ( 'Re-run the command with --passwd-len={}' if pf in ('bip39','hex') else
'Re-run the command, specifying a password length of {} or less' )
die(1,(m1+'\n'+m2).format(len(seed.data) * 8,good_pw_len))
die(1,
'Cannot generate passwords with more entropy than underlying seed! ({} bits)\n'.format(
len(seed.data) * 8 ) + (
'Re-run the command with --passwd-len={}' if pf in ('bip39','hex') else
'Re-run the command, specifying a password length of {} or less'
).format(good_pw_len) )
if pf in ('bip39','hex') and pw_bytes < seed.byte_len:
m1 = 'WARNING: requested {} length has less entropy than underlying seed!'
m2 = 'Is this what you want?'
if not keypress_confirm((m1+'\n'+m2).format(self.pw_info[pf].desc),default_yes=True):
if not keypress_confirm(
f'WARNING: requested {self.pw_info[pf].desc} length has less entropy ' +
'than underlying seed!\nIs this what you want?',
default_yes = True ):
die(1,'Exiting at user request')
def make_passwd(self,hex_sec):
@ -1038,22 +1048,22 @@ Record this checksum: it will be used to verify the password file in the future
def check_format(self,pw):
if not self.pw_info[self.pw_fmt].chk_func(pw):
raise ValueError('Password is not valid {} data'.format(self.pw_info[self.pw_fmt].desc))
raise ValueError(f'Password is not valid {self.pw_info[self.pw_fmt].desc} data')
pwlen = len(pw.split()) if self.pw_fmt in ('bip39','xmrseed') else len(pw)
if pwlen != self.pw_len:
raise ValueError('Password has incorrect length ({} != {})'.format(pwlen,self.pw_len))
raise ValueError(f'Password has incorrect length ({pwlen} != {self.pw_len})')
return True
def scramble_seed(self,seed):
# Changing either pw_fmt or pw_len will cause a different, unrelated
# set of passwords to be generated: this is what we want.
# NB: In original implementation, pw_id_str was 'baseN', not 'bN'
scramble_key = '{}:{}:{}'.format(self.pw_fmt,self.pw_len,self.pw_id_str)
scramble_key = f'{self.pw_fmt}:{self.pw_len}:{self.pw_id_str}'
if self.hex2bip39:
from .bip39 import bip39
pwlen = bip39.nwords2seedlen(self.pw_len,in_hex=True)
scramble_key = '{}:{}:{}'.format('hex',pwlen,self.pw_id_str)
scramble_key = f'hex:{pwlen}:{self.pw_id_str}'
from .crypto import scramble_seed
dmsg_sc('str',scramble_key)
@ -1064,11 +1074,11 @@ Record this checksum: it will be used to verify the password file in the future
if self.pw_fmt in ('bip39','xmrseed'):
ret = lines.pop(0).split(None,self.pw_len+1)
if len(ret) > self.pw_len+1:
m1 = 'extraneous text {!r} found after password'.format(ret[self.pw_len+1])
m1 = f'extraneous text {ret[self.pw_len+1]!r} found after password'
m2 = '[bare comments not allowed in BIP39 password files]'
m = m1+' '+m2
elif len(ret) < self.pw_len+1:
m = 'invalid password length {}'.format(len(ret)-1)
m = f'invalid password length {len(ret)-1}'
else:
return (ret[0],' '.join(ret[1:self.pw_len+1]),'')
raise ValueError(m)
@ -1077,16 +1087,16 @@ Record this checksum: it will be used to verify the password file in the future
return ret if len(ret) == 3 else ret + ['']
def make_label(self):
return '{} {} {}:{}'.format(self.al_id.sid,self.pw_id_str,self.pw_fmt_disp,self.pw_len)
return f'{self.al_id.sid} {self.pw_id_str} {self.pw_fmt_disp}:{self.pw_len}'
class AddrData(MMGenObject):
msgs = {
'too_many_acct_addresses': """
'too_many_acct_addresses': f"""
ERROR: More than one address found for account: '{{}}'.
Your 'wallet.dat' file appears to have been altered by a non-{pnm} program.
Please restore your tracking wallet from a backup or create a new one and
re-import your addresses.
""".strip().format(pnm=pnm)
""".strip()
}
def __new__(cls,proto,*args,**kwargs):
@ -1121,7 +1131,7 @@ re-import your addresses.
self.al_ids[addrlist.al_id] = addrlist
return True
else:
raise TypeError('Error: object {!r} is not of type AddrList'.format(addrlist))
raise TypeError(f'Error: object {addrlist!r} is not of type AddrList')
def make_reverse_dict(self,coinaddrs):
d = MMGenDict()
@ -1169,6 +1179,6 @@ class TwAddrData(AddrData,metaclass=AsyncInit):
out[al_id].append(AddrListEntry(self.proto,idx=obj.idx,addr=addr_array[0],label=l.comment))
i += 1
vmsg('{n} {pnm} addresses found, {m} accounts total'.format(n=i,pnm=pnm,m=len(twd)))
vmsg(f'{i} {pnm} addresses found, {len(twd)} accounts total')
for al_id in out:
self.add(AddrList(self.proto,al_id=al_id,adata=AddrListData(sorted(out[al_id],key=lambda a: a.idx))))

53
mmgen/altcoin.py
View file

@ -36,7 +36,9 @@ altcoin.py - Coin constants for Bitcoin-derived altcoins
# NBT: 150/191 c/u, 25/('B'), 26/('B')
import sys
def msg(s): sys.stderr.write(s+'\n')
def msg(s):
sys.stderr.write(s+'\n')
def test_equal(desc,a,b,*cdata):
if type(a) == int:
@ -44,11 +46,16 @@ def test_equal(desc,a,b,*cdata):
b = hex(b)
(network,coin,e,b_desc,verbose) = cdata
if verbose:
m = ' {:20}: {!r}'
msg(m.format(desc,a))
msg(f' {desc:20}: {a!r}')
if a != b:
m = '{}s for {} {} do not match:\n CoinInfo: {}\n {}: {}'
raise ValueError(m.format(desc.capitalize(),coin.upper(),network,a,b_desc,b))
raise ValueError(
'{}s for {} {} do not match:\n CoinInfo: {}\n {}: {}'.format(
desc.capitalize(),
coin.upper(),
network,
a,
b_desc,
b ))
from collections import namedtuple
ce = namedtuple('CoinInfoEntry',
@ -415,7 +422,7 @@ class CoinInfo(object):
cdata = (network,coin,e,'Computed value',verbose)
if not quiet:
msg('{} {}'.format(coin,network))
msg(f'{coin} {network}')
vn_info = e.p2pkh_info
ret = cls.find_addr_leading_symbol(vn_info[0])
@ -437,7 +444,7 @@ class CoinInfo(object):
proto = init_proto(coin,testnet=network=='testnet')
cdata = (network,coin,e,type(proto).__name__,verbose)
if not quiet:
msg('Verifying {} {}'.format(coin.upper(),network))
msg(f'Verifying {coin.upper()} {network}')
if coin != 'bch': # TODO
test_equal('coin name',e.name,proto.name,*cdata)
@ -490,9 +497,9 @@ class CoinInfo(object):
e[k] = list(e[k])
e[k][0] = myhex(e[k][0])
s1 = cls.find_addr_leading_symbol(int(e[k][0][2:],16))
m = 'Fixing leading address letter for coin {} ({!r} --> {})'.format(e['symbol'],e[k][1],s1)
m = f'Fixing leading address letter for coin {e["symbol"]} ({e[k][1]!r} --> {s1})'
if e[k][1] != '?':
assert s1 == e[k][1],'First letters do not match! {}'.format(m)
assert s1 == e[k][1], f'First letters do not match! {m}'
else:
msg(m)
e[k][1] = s1
@ -501,7 +508,7 @@ class CoinInfo(object):
old_sym = None
for sym in sorted([e.symbol for e in data]):
if sym == old_sym:
msg("'{}': duplicate coin symbol in data!".format(sym))
msg(f'{sym!r}: duplicate coin symbol in data!')
sys.exit(2)
old_sym = sym
@ -528,20 +535,20 @@ class CoinInfo(object):
if sym in tt:
src = tt[sym]
if src != trust:
msg("Updating trust for coin '{}': {} -> {}".format(sym,trust,src))
msg(f'Updating trust for coin {sym!r}: {trust} -> {src}')
e['trust_level'] = src
else:
if trust != 0:
msg("Downgrading trust for coin '{}': {} -> {}".format(sym,trust,0))
msg(f'Downgrading trust for coin {sym!r}: {trust} -> 0')
e['trust_level'] = 0
if sym in cls.cross_checks:
if int(e['trust_level']) == 0 and len(cls.cross_checks[sym]) > 1:
msg("Upgrading trust for coin '{}': {} -> {}".format(sym,e['trust_level'],1))
msg(f'Upgrading trust for coin {sym!r}: {e["trust_level"]} -> 1')
e['trust_level'] = 1
print(fs.format(*e.values()))
msg('Processed {} entries'.format(len(data)))
msg(f'Processed {len(data)} entries')
@classmethod
def find_addr_leading_symbol(cls,ver_num,verbose=False):
@ -569,17 +576,17 @@ class CoinInfo(object):
def print_symbols(cls,include_names=False,reverse=False):
for e in cls.coin_constants['mainnet']:
if reverse:
print('{:6} {}'.format(e.symbol,e.name))
print(f'{e.symbol:6} {e.name}')
else:
name_w = max(len(e.name) for e in cls.coin_constants['mainnet'])
print(('{:{}} '.format(e.name,name_w) if include_names else '') + e.symbol)
print((f'{e.name:{name_w}} ' if include_names else '') + e.symbol)
@classmethod
def create_trust_table(cls):
tt = {}
mn = cls.external_tests['mainnet']
for ext_prog in mn:
assert len(set(mn[ext_prog])) == len(mn[ext_prog]),"Duplicate entry in '{}'!".format(ext_prog)
assert len(set(mn[ext_prog])) == len(mn[ext_prog]), f'Duplicate entry in {ext_prog!r}!'
for coin in mn[ext_prog]:
if coin in tt:
tt[coin] += 1
@ -609,8 +616,7 @@ class CoinInfo(object):
if verbose:
m1 = 'Requested tool {t!r} does not support coin {c} on network {n}'
m2 = 'No test tool found for coin {c} on network {n}'
m = m1 if tool_arg else m2
msg(m.format(t=tool,c=coin,n=network))
msg((m1 if tool_arg else m2).format(t=tool,c=coin,n=network))
return None
if addr_type == 'zcash_z':
@ -618,8 +624,7 @@ class CoinInfo(object):
return 'zcash-mini'
else:
if verbose:
m = "Address type {a!r} supported only by tool 'zcash-mini'"
msg(m.format(a=addr_type))
msg(f"Address type {addr_type!r} supported only by tool 'zcash-mini'")
return None
try:
@ -630,8 +635,7 @@ class CoinInfo(object):
pass
else:
if verbose:
m = 'Tool {t!r} blacklisted for coin {c}, addr_type {a!r}'
msg(m.format(t=tool,c=coin,a=addr_type))
msg(f'Tool {tool!r} blacklisted for coin {coin}, addr_type {addr_type!r}')
return None
if tool_arg: # skip whitelists
@ -645,8 +649,7 @@ class CoinInfo(object):
if verbose:
m1 = 'Requested tool {t!r} does not support coin {c}, addr_type {a!r}, on network {n}'
m2 = 'No test tool found supporting coin {c}, addr_type {a!r}, on network {n}'
m = m1 if tool_arg else m2
msg(m.format(t=tool,c=coin,n=network,a=addr_type))
msg((m1 if tool_arg else m2).format(t=tool,c=coin,n=network,a=addr_type))
return None
return tool

23
mmgen/altcoins/eth/contract.py
View file

@ -51,7 +51,9 @@ class TokenBase(MMGenObject): # ERC20
async def do_call(self,method_sig,method_args='',toUnit=False):
data = create_method_id(method_sig) + method_args
if g.debug:
msg('ETH_CALL {}: {}'.format(method_sig,'\n '.join(parse_abi(data))))
msg('ETH_CALL {}: {}'.format(
method_sig,
'\n '.join(parse_abi(data)) ))
ret = await self.rpc.call('eth_call',{ 'to': '0x'+self.addr, 'data': '0x'+data },'pending')
if self.proto.network == 'regtest' and g.daemon_id == 'erigon': # ERIGON
import asyncio
@ -79,19 +81,19 @@ class TokenBase(MMGenObject): # ERC20
assert ret[:2] == '0x'
return int(ret,16)
except:
msg("RPC call to decimals() failed (returned '{}')".format(ret))
msg(f'RPC call to decimals() failed (returned {ret!r})')
return None
async def get_total_supply(self):
return await self.do_call('totalSupply()',toUnit=True)
async def info(self):
fs = '{:15}{}\n' * 5
return fs.format('token address:', self.addr,
'token symbol:', await self.get_symbol(),
'token name:', await self.get_name(),
'decimals:', self.decimals,
'total supply:', await self.get_total_supply())
return ('{:15}{}\n' * 5).format(
'token address:', self.addr,
'token symbol:', await self.get_symbol(),
'token name:', await self.get_name(),
'decimals:', self.decimals,
'total supply:', await self.get_total_supply() )
async def code(self):
return (await self.rpc.call('eth_getCode','0x'+self.addr))[2:]
@ -99,7 +101,7 @@ class TokenBase(MMGenObject): # ERC20
def create_data(self,to_addr,amt,method_sig='transfer(address,uint256)',from_addr=None):
from_arg = from_addr.rjust(64,'0') if from_addr else ''
to_arg = to_addr.rjust(64,'0')
amt_arg = '{:064x}'.format(int(amt / self.base_unit))
amt_arg = '{:064x}'.format( int(amt / self.base_unit) )
return create_method_id(method_sig) + from_arg + to_arg + amt_arg
def make_tx_in( self,from_addr,to_addr,amt,start_gas,gasPrice,nonce,
@ -128,7 +130,8 @@ class TokenBase(MMGenObject): # ERC20
if g.debug:
msg('TOKEN DATA:')
pp_msg(tx.to_dict())
msg('PARSED ABI DATA:\n {}'.format('\n '.join(parse_abi(tx.data.hex()))))
msg('PARSED ABI DATA:\n {}'.format(
'\n '.join(parse_abi(tx.data.hex())) ))
return hex_tx,coin_txid
# The following are used for token deployment only:

6
mmgen/altcoins/eth/tw.py
View file

@ -43,7 +43,7 @@ class EthereumTrackingWallet(TrackingWallet):
upgraded = False
if not 'accounts' in self.data or not 'coin' in self.data:
ymsg('Upgrading {} (v1->v2: accounts field added)'.format(self.desc))
ymsg(f'Upgrading {self.desc} (v1->v2: accounts field added)')
if not 'accounts' in self.data:
self.data = {}
import json
@ -66,13 +66,13 @@ class EthereumTrackingWallet(TrackingWallet):
upgraded = True
if self.data['tokens'] and not have_token_params_fields():
ymsg('Upgrading {} (v2->v3: token params fields added)'.format(self.desc))
ymsg(f'Upgrading {self.desc} (v2->v3: token params fields added)')
add_token_params_fields()
upgraded = True
if upgraded:
self.force_write()
msg('{} upgraded successfully!'.format(self.desc))
msg(f'{self.desc} upgraded successfully!')
async def rpc_get_balance(self,addr):
return ETHAmt(int(await self.rpc.call('eth_getBalance','0x'+addr,'latest'),16),'wei')

35
mmgen/altcoins/eth/tx.py
View file

@ -46,7 +46,7 @@ class EthereumMMGenTX:
# given absolute fee in ETH, return gas price in Gwei using tx_gas
def fee_abs2rel(self,abs_fee,to_unit='Gwei'):
ret = ETHAmt(int(abs_fee.toWei() // self.tx_gas.toWei()),'wei')
dmsg('fee_abs2rel() ==> {} ETH'.format(ret))
dmsg(f'fee_abs2rel() ==> {ret} ETH')
return ret if to_unit == 'eth' else ret.to_unit(to_unit,show_decimal=True)
def get_hex_locktime(self):
@ -54,7 +54,7 @@ class EthereumMMGenTX:
# given rel fee (gasPrice) in wei, return absolute fee using tx_gas (not in MMGenTX)
def fee_gasPrice2abs(self,rel_fee):
assert isinstance(rel_fee,int),"'{}': incorrect type for fee estimate (not an integer)".format(rel_fee)
assert isinstance(rel_fee,int), f'{rel_fee!r}: incorrect type for fee estimate (not an integer)'
return ETHAmt(rel_fee * self.tx_gas.toWei(),'wei')
def is_replaceable(self):
@ -114,7 +114,7 @@ class EthereumMMGenTX:
assert len(self.inputs) == 1,'Transaction has more than one input!'
o_num = len(self.outputs)
o_ok = 0 if self.usr_contract_data else 1
assert o_num == o_ok,'Transaction has {} output{} (should have {})'.format(o_num,suf(o_num),o_ok)
assert o_num == o_ok, f'Transaction has {o_num} output{suf(o_num)} (should have {o_ok})'
await self.make_txobj()
odict = { k: str(v) for k,v in self.txobj.items() if k != 'token_to' }
self.hex = json.dumps(odict)
@ -147,7 +147,7 @@ class EthereumMMGenTX:
elif int(reply) < 1:
msg('Account number must be >= 1')
elif int(reply) > len(unspent):
msg('Account number must be <= {}'.format(len(unspent)))
msg(f'Account number must be <= {len(unspent)}')
else:
return [int(reply)]
@ -175,7 +175,7 @@ class EthereumMMGenTX:
def fee_est2abs(self,rel_fee,fe_type=None):
ret = self.fee_gasPrice2abs(rel_fee) * opt.tx_fee_adj
if opt.verbose:
msg('Estimated fee: {} ETH'.format(ret))
msg(f'Estimated fee: {ret} ETH')
return ret
def convert_and_check_fee(self,tx_fee,desc='Missing description'):
@ -183,8 +183,11 @@ class EthereumMMGenTX:
if abs_fee == False:
return False
elif not self.disable_fee_check and (abs_fee > self.proto.max_tx_fee):
m = '{} {c}: {} fee too large (maximum fee: {} {c})'
msg(m.format(abs_fee.hl(),desc,self.proto.max_tx_fee.hl(),c=self.proto.coin))
msg('{} {c}: {} fee too large (maximum fee: {} {c})'.format(
abs_fee.hl(),
desc,
self.proto.max_tx_fee.hl(),
c = self.proto.coin ))
return False
else:
return abs_fee
@ -211,12 +214,12 @@ class EthereumMMGenTX:
raise UserAddressNotInWallet(m.format(i))
ret.append(i)
else:
die(1,"'{}': not an MMGen ID or coin address".format(i))
die(1,f'{i!r}: not an MMGen ID or coin address')
return ret
def final_inputs_ok_msg(self,funds_left):
chg = '0' if (self.outputs and self.outputs[0].is_chg) else funds_left
return "Transaction leaves {} {} in the sender's account".format(
return 'Transaction leaves {} {} in the sender’s account'.format(
ETHAmt(chg).hl(),
self.proto.coin
)
@ -426,10 +429,10 @@ class EthereumMMGenTX:
if self.txobj['data']:
cd = capfirst(self.contract_desc)
if r.exec_status == 0:
msg('{} failed to execute!'.format(cd))
msg(f'{cd} failed to execute!')
else:
msg('{} successfully executed with status {}'.format(cd,r.exec_status))
die(0,'Transaction has {} confirmation{}'.format(r.confs,suf(r.confs)))
msg(f'{cd} successfully executed with status {r.exec_status}')
die(0,f'Transaction has {r.confs} confirmation{suf(r.confs)}')
die(1,'Transaction is neither in mempool nor blockchain!')
async def send(self,prompt_user=True,exit_on_fail=False):
@ -458,7 +461,7 @@ class EthereumMMGenTX:
ret = False
if ret == False:
msg(red('Send of MMGen transaction {} failed'.format(self.txid)))
msg(red(f'Send of MMGen transaction {self.txid} failed'))
if exit_on_fail:
sys.exit(1)
return False
@ -553,9 +556,9 @@ class EthereumTokenMMGenTX:
def format_view_body(self,*args,**kwargs):
return 'Token: {d} {c}\n{r}'.format(
d=self.txobj['token_addr'].hl(),
c=blue('(' + self.proto.dcoin + ')'),
r=super().format_view_body(*args,**kwargs))
d = self.txobj['token_addr'].hl(),
c = blue('(' + self.proto.dcoin + ')'),
r = super().format_view_body(*args,**kwargs ))
class Unsigned(Completed,EthereumMMGenTX.Unsigned):
desc = 'unsigned transaction'

37
mmgen/baseconv.py
View file

@ -85,7 +85,7 @@ class baseconv(object):
from .mn_tirosh import words
cls.digits[mn_id] = words[:cls.mn_base]
else:
raise ValueError('{}: unrecognized mnemonic ID'.format(mn_id))
raise ValueError(f'{mn_id}: unrecognized mnemonic ID')
@classmethod
def get_wordlist(cls,wl_id):
@ -101,7 +101,7 @@ class baseconv(object):
def check_wordlists(cls):
for k,v in list(cls.wl_chksums.items()):
res = cls.get_wordlist_chksum(k)
assert res == v,'{}: checksum mismatch for {} (should be {})'.format(res,k,v)
assert res == v,f'{res}: checksum mismatch for {k} (should be {v})'
return True
@classmethod
@ -110,7 +110,7 @@ class baseconv(object):
wl = cls.digits[wl_id]
from .util import qmsg,compare_chksums
ret = 'Wordlist: {}\nLength: {} words'.format(wl_id,len(wl))
ret = f'Wordlist: {wl_id}\nLength: {len(wl)} words'
new_chksum = cls.get_wordlist_chksum(wl_id)
a,b = 'generated','saved'
@ -136,8 +136,7 @@ class baseconv(object):
elif pad == 'seed':
return seed_pad_func()
else:
m = "{!r}: illegal value for 'pad' (must be None,'seed' or int)"
raise BaseConversionPadError(m.format(pad))
raise BaseConversionPadError(f"{pad!r}: illegal value for 'pad' (must be None,'seed' or int)")
@staticmethod
def monero_mn_checksum(words):
@ -161,14 +160,14 @@ class baseconv(object):
desc = cls.desc[wl_id][0]
if len(words) == 0:
raise BaseConversionError('empty {} data'.format(desc))
raise BaseConversionError(f'empty {desc} data')
def get_seed_pad():
assert wl_id in cls.seedlen_map_rev,'seed padding not supported for base {!r}'.format(wl_id)
assert wl_id in cls.seedlen_map_rev,f'seed padding not supported for base {wl_id!r}'
d = cls.seedlen_map_rev[wl_id]
if not len(words) in d:
m = '{}: invalid length for seed-padded {} data in base conversion'
raise BaseConversionError(m.format(len(words),desc))
raise BaseConversionError(
f'{len(words)}: invalid length for seed-padded {desc} data in base conversion' )
return d[len(words)]
pad_val = max(cls.get_pad(pad,get_seed_pad),1)
@ -176,12 +175,13 @@ class baseconv(object):
base = len(wl)
if not set(words) <= set(wl):
m = ('{w!r}:','seed data')[pad=='seed'] + ' not in {d} format'
raise BaseConversionError(m.format(w=words_arg,d=desc))
raise BaseConversionError(
( 'seed data' if pad == 'seed' else f'{words_arg!r}:' ) +
f' not in {desc} format' )
if wl_id == 'xmrseed':
if len(words) not in cls.seedlen_map_rev['xmrseed']:
die(2,'{}: invalid length for Monero mnemonic'.format(len(words)))
die(2,f'{len(words)}: invalid length for Monero mnemonic')
z = cls.monero_mn_checksum(words[:-1])
assert z == words[-1],'invalid Monero mnemonic checksum'
@ -204,8 +204,9 @@ class baseconv(object):
from .util import is_hex_str
if not is_hex_str(hexstr):
m = ('{h!r}:','seed data')[pad=='seed'] + ' not a hexadecimal string'
raise HexadecimalStringError(m.format(h=hexstr))
raise HexadecimalStringError(
( 'seed data' if pad == 'seed' else f'{hexstr!r}:' ) +
' not a hexadecimal string' )
return cls.frombytes(bytes.fromhex(hexstr),wl_id,pad,tostr)
@ -220,11 +221,11 @@ class baseconv(object):
raise BaseConversionError('empty data not allowed in base conversion')
def get_seed_pad():
assert wl_id in cls.seedlen_map,'seed padding not supported for base {!r}'.format(wl_id)
assert wl_id in cls.seedlen_map, f'seed padding not supported for base {wl_id!r}'
d = cls.seedlen_map[wl_id]
if not len(bytestr) in d:
m = '{}: invalid byte length for seed data in seed-padded base conversion'
raise SeedLengthError(m.format(len(bytestr)))
raise SeedLengthError(
f'{len(bytestr)}: invalid byte length for seed data in seed-padded base conversion' )
return d[len(bytestr)]
pad = max(cls.get_pad(pad,get_seed_pad),1)
@ -233,7 +234,7 @@ class baseconv(object):
if wl_id == 'xmrseed':
if len(bytestr) not in cls.seedlen_map['xmrseed']:
die(2,'{}: invalid seed byte length for Monero mnemonic'.format(len(bytestr)))
die(2, f'{len(bytestr)}: invalid seed byte length for Monero mnemonic')
def num2base_monero(num):
w1 = num % base

14
mmgen/bip39.py
View file

@ -2099,7 +2099,7 @@ zoo
for k,v in cls.constants.items():
if v[1] == nwords:
return int(k)//8 if in_bytes else int(k)//4 if in_hex else int(k)
raise MnemonicError('{!r}: invalid word length for BIP39 mnemonic'.format(nwords))
raise MnemonicError(f'{nwords!r}: invalid word length for BIP39 mnemonic')
@classmethod
def seedlen2nwords(cls,seed_len,in_bytes=False,in_hex=False):
@ -2107,7 +2107,7 @@ zoo
try:
return cls.constants[str(seed_bits)][1]
except:
raise ValueError('{!r}: invalid seed length for BIP39 mnemonic'.format(seed_bits))
raise ValueError(f'{seed_bits!r}: invalid seed length for BIP39 mnemonic')
@classmethod
def tohex(cls,words,wl_id,pad=None):
@ -2118,7 +2118,7 @@ zoo
for n,w in enumerate(words):
if w not in wl:
raise MnemonicError('word #{} is not in the BIP39 word list'.format(n+1))
raise MnemonicError(f'word #{n+1} is not in the BIP39 word list')
res = ''.join(['{:011b}'.format(wl.index(w)) for w in words])
@ -2127,10 +2127,10 @@ zoo
bitlen = int(k)
break
else:
raise MnemonicError('{}: invalid BIP39 seed phrase length'.format(len(words)))
raise MnemonicError(f'{len(words)}: invalid BIP39 seed phrase length')
if pad != None:
assert pad * 4 == bitlen, '{}: invalid pad length'.format(pad)
assert pad * 4 == bitlen, f'{pad}: invalid pad length'
seed_bin = res[:bitlen]
chk_bin = res[bitlen:]
@ -2157,11 +2157,11 @@ zoo
seed_bytes = bytes.fromhex(seed_hex)
bitlen = len(seed_bytes) * 8
assert str(bitlen) in cls.constants,'{}: invalid seed bit length'.format(bitlen)
assert str(bitlen) in cls.constants, f'{bitlen}: invalid seed bit length'
chk_len,mn_len = cls.constants[str(bitlen)]
if pad != None:
assert mn_len == pad, '{}: invalid pad length'.format(pad)
assert mn_len == pad, f'{pad}: invalid pad length'
chk_hex = sha256(seed_bytes).hexdigest()

34
mmgen/cfg.py
View file

@ -56,7 +56,7 @@ class CfgFile(object):
self.data = ''
def copy_data(self):
assert self.write_ok, 'writing to file {!r} not allowed!'.format(self.fn)
assert self.write_ok, f'writing to file {self.fn!r} not allowed!'
src = cfg_file('sys')
if src.data:
data = src.data + src.make_metadata() if self.write_metadata else src.data
@ -64,7 +64,7 @@ class CfgFile(object):
open(self.fn,'w').write('\n'.join(data)+'\n')
os.chmod(self.fn,0o600)
except:
die(2,'ERROR: unable to write to {!r}'.format(self.fn))
die(2,f'ERROR: unable to write to {self.fn!r}')
def parse_value(self,value,refval):
if isinstance(refval,dict):
@ -89,7 +89,7 @@ class CfgFile(object):
if m:
yield self.line_data(m[1],m[3],lineno,None)
else:
raise CfgFileParseError('Parse error in file {!r}, line {}'.format(self.fn,lineno))
raise CfgFileParseError(f'Parse error in file {self.fn!r}, line {lineno}')
return gen_lines()
@classmethod
@ -105,7 +105,7 @@ class CfgFileSample(CfgFile):
@classmethod
def cls_make_metadata(cls,data):
return ['# Version {} {}'.format(cls.cur_ver,cls.compute_chksum(data))]
return [f'# Version {cls.cur_ver} {cls.compute_chksum(data)}']
@staticmethod
def compute_chksum(data):
@ -131,7 +131,7 @@ class CfgFileSample(CfgFile):
if m:
return self.line_data(m[2],m[4],lineno,chunk)
else:
raise CfgFileParseError('Parse error in file {!r}, line {}'.format(self.fn,lineno))
raise CfgFileParseError(f'Parse error in file {self.fn!r}, line {lineno}')
def gen_chunks(lines):
hdr = True
@ -159,7 +159,7 @@ class CfgFileSample(CfgFile):
chunk.append(line)
last_nonblank = lineno
else:
raise CfgFileParseError('Parse error in file {!r}, line {}'.format(self.fn,lineno))
raise CfgFileParseError(f'Parse error in file {self.fn!r}, line {lineno}')
if chunk:
yield process_chunk(chunk,last_nonblank)
@ -201,7 +201,7 @@ class CfgFileSampleSys(CfgFileSample):
self.data = files('mmgen').joinpath('data',self.fn_base).read_text().splitlines()
def make_metadata(self):
return ['# Version {} {}'.format(self.cur_ver,self.computed_chksum)]
return [f'# Version {self.cur_ver} {self.computed_chksum}']
class CfgFileSampleUsr(CfgFileSample):
desc = 'sample configuration file'
@ -260,25 +260,23 @@ class CfgFileSampleUsr(CfgFileSample):
def show_changes(self,diff):
ymsg('Warning: configuration file options have changed!\n')
m1 = ' The following option{} been {}:\n {}\n'
m2 = """
The following removed option{} set in {!r}
and must be deleted or commented out:
{}
"""
for desc in ('added','removed'):
data = diff[desc]
if data:
opts = fmt_list([i.name for i in data],fmt='bare')
msg(m1.format(suf(data,verb='has'),desc,opts))
msg(f' The following option{suf(data,verb="has")} been {desc}:\n {opts}\n')
if desc == 'removed' and data:
uc = cfg_file('usr')
usr_names = [i.name for i in uc.get_lines()]
rm_names = [i.name for i in data]
bad = sorted(set(usr_names).intersection(rm_names))
if bad:
ymsg(fmt(m2,' ').format(suf(bad,verb='is'),uc.fn,' '+fmt_list(bad,fmt='bare')))
m = f"""
The following removed option{suf(bad,verb='is')} set in {uc.fn!r}
and must be deleted or commented out:
{' ' + fmt_list(bad,fmt='bare')}
"""
ymsg(fmt(m,indent=' ',strip_char='\t'))
while True:
if not keypress_confirm(self.details_confirm_prompt,no_nl=True):
return
@ -288,9 +286,9 @@ class CfgFileSampleUsr(CfgFileSample):
sep,sep2 = ('\n ','\n\n ')
if data:
yield (
'{} section{}:'.format(capfirst(desc),suf(data))
f'{capfirst(desc)} section{suf(data)}:'
+ sep2
+ sep2.join(['{}'.format(sep.join(v.chunk)) for v in data])
+ sep2.join([f'{sep.join(v.chunk)}' for v in data])
)
do_pager(

43
mmgen/crypto.py
View file

@ -34,8 +34,7 @@ def scramble_seed(seed,scramble_key):
import hmac
step1 = hmac.new(seed,scramble_key,sha256).digest()
if g.debug:
fs = 'Seed: {!r}\nScramble key: {}\nScrambled seed: {}\n'
msg(fs.format(seed.hex(),scramble_key,step1.hex()))
msg(f'Seed: {seed.hex()!r}\nScramble key: {scramble_key}\nScrambled seed: {step1.hex()}\n')
return sha256_rounds(step1,g.scramble_hash_rounds)
def encrypt_seed(seed,key):
@ -64,30 +63,30 @@ def decrypt_seed(enc_seed,key,seed_id,key_id):
vmsg('')
return False
# else:
# qmsg('Generated IDs (Seed/Key): {}/{}'.format(chk2,chk1))
# qmsg(f'Generated IDs (Seed/Key): {chk2}/{chk1}')
dmsg('Decrypted seed: {}'.format(dec_seed.hex()))
dmsg(f'Decrypted seed: {dec_seed.hex()}')
return dec_seed
def encrypt_data(data,key,iv=g.aesctr_dfl_iv,desc='data',verify=True):
vmsg('Encrypting {}'.format(desc))
vmsg(f'Encrypting {desc}')
c = Cipher(algorithms.AES(key),modes.CTR(iv),backend=default_backend())
encryptor = c.encryptor()
enc_data = encryptor.update(data) + encryptor.finalize()
if verify:
vmsg_r('Performing a test decryption of the {}...'.format(desc))
vmsg_r(f'Performing a test decryption of the {desc}...')
c = Cipher(algorithms.AES(key),modes.CTR(iv),backend=default_backend())
encryptor = c.encryptor()
dec_data = encryptor.update(enc_data) + encryptor.finalize()
if dec_data != data:
die(2,"ERROR.\nDecrypted {s} doesn't match original {s}".format(s=desc))
die(2,f'ERROR.\nDecrypted {desc} doesn’t match original {desc}')
vmsg('done')
return enc_data
def decrypt_data(enc_data,key,iv=g.aesctr_dfl_iv,desc='data'):
vmsg_r('Decrypting {} with key...'.format(desc))
vmsg_r(f'Decrypting {desc} with key...')
c = Cipher(algorithms.AES(key),modes.CTR(iv),backend=default_backend())
encryptor = c.encryptor()
return encryptor.update(enc_data) + encryptor.finalize()
@ -136,12 +135,11 @@ def scrypt_hash_passphrase(passwd,salt,hash_preset,buflen=32):
return ret
def make_key(passwd,salt,hash_preset,desc='encryption key',from_what='passphrase',verbose=False):
if from_what: desc += ' from '
if opt.verbose or verbose:
msg_r('Generating {}{}...'.format(desc,from_what))
msg_r(f"Generating {desc}{' from ' + from_what if from_what else ''}...")
key = scrypt_hash_passphrase(passwd,salt,hash_preset)
if opt.verbose or verbose: msg('done')
dmsg('Key: {}'.format(key.hex()))
dmsg(f'Key: {key.hex()}')
return key
def _get_random_data_from_user(uchars,desc):
@ -166,9 +164,8 @@ def _get_random_data_from_user(uchars,desc):
"""
msg(f'Enter {uchars} random symbols' if opt.quiet else
'\n{}\n{}'.format( fmt(info1,indent=' '), fmt(info2) ))
prompt = 'You may begin typing. {} symbols left: '
'\n' + fmt(info1,indent=' ') +
'\n' + fmt(info2) )
import time
from .term import get_char_raw
@ -176,12 +173,12 @@ def _get_random_data_from_user(uchars,desc):
time_data = []
for i in range(uchars):
key_data += get_char_raw('\r'+prompt.format(uchars-i))
key_data += get_char_raw(f'\rYou may begin typing. {uchars-i} symbols left: ')
time_data.append(time.time())
msg_r( '\r' if opt.quiet else f"\rThank you. That's enough.{' '*18}\n\n" )
msg_r( '\r' if opt.quiet else f'\rThank you. That’s enough.{" "*18}\n\n' )
time_data = ['{:.22f}'.format(t).rstrip('0') for t in time_data]
time_data = [f'{t:.22f}'.rstrip('0') for t in time_data]
avg_prec = sum(len(t.split('.')[1]) for t in time_data) // len(time_data)
@ -217,21 +214,21 @@ def add_user_random(rand_bytes,desc):
return rand_bytes
def get_hash_preset_from_user(hp=g.dfl_hash_preset,desc='data'):
prompt = f'Enter hash preset for {desc},\nor hit ENTER to accept the default value ({hp!r}): '
while True:
ret = my_raw_input(prompt)
ret = my_raw_input(
f'Enter hash preset for {desc},\n' +
f'or hit ENTER to accept the default value ({hp!r}): ' )
if ret:
if ret in g.hash_presets:
return ret
else:
msg('Invalid input. Valid choices are {}'.format(', '.join(g.hash_presets)))
msg(f'Invalid input. Valid choices are {", ".join(g.hash_presets)}')
continue
else:
return hp
def get_new_passphrase(desc,passchg=False):
pw_desc = '{}passphrase for {}'.format(('','new ')[bool(passchg)], desc)
pw_desc = f"{'new ' if passchg else ''}passphrase for {desc}"
if opt.passwd_file:
pw = ' '.join(get_words_from_file(opt.passwd_file,pw_desc))
elif opt.echo_passphrase:
@ -253,7 +250,7 @@ def get_new_passphrase(desc,passchg=False):
return pw
def get_passphrase(desc,passchg=False):
pw_desc ='{}passphrase for {}'.format(('','old ')[bool(passchg)],desc)
pw_desc = f"{'old ' if passchg else ''}passphrase for {desc}"
if opt.passwd_file:
pwfile_reuse_warning(opt.passwd_file)
return ' '.join(get_words_from_file(opt.passwd_file,pw_desc))

4
mmgen/daemon.py
View file

@ -88,7 +88,7 @@ class Daemon(Lockable):
msg(f'Starting {self.desc} on port {self.bind_port}')
if self.debug:
msg('\nExecuting: {}'.format(' '.join(cmd)))
msg(f'\nExecuting: {" ".join(cmd)}')
if self.use_threads and is_daemon and not self.opt.no_daemonize:
ret = self.exec_cmd_thread(cmd)
@ -445,7 +445,7 @@ class CoinDaemon(Daemon):
os.makedirs(self.datadir,exist_ok=True)
if self.cfg_file and not self.flag.keep_cfg_file:
open('{}/{}'.format(self.datadir,self.cfg_file),'w').write(self.cfg_file_hdr)
open(f'{self.datadir}/{self.cfg_file}','w').write(self.cfg_file_hdr)
if self.use_pidfile and os.path.exists(self.pidfile):
# Parity overwrites the data in the existing pidfile without zeroing it first, leading

44
mmgen/devtools.py
View file

@ -64,24 +64,27 @@ if os.getenv('MMGEN_DEBUG') or os.getenv('MMGEN_TEST_SUITE') or os.getenv('MMGEN
if is_dict:
out.append('{s}{:<{l}}'.format(i,s=' '*(4*lvl+8),l=10,l2=8*(lvl+1)+8))
if hasattr(el,'pfmt'):
out.append('{:>{l}}{}'.format('',el.pfmt(
lvl=lvl+1,id_list=id_list+[id(self)]),l=(lvl+1)*8))
out.append('{:>{l}}{}'.format(
'',
el.pfmt( lvl=lvl+1, id_list=id_list+[id(self)] ),
l = (lvl+1)*8 ))
elif isinstance(el,scalars):
if isList(e):
out.append('{:>{l}}{:16}\n'.format('',repr(el),l=lvl*8))
out.append( '{:>{l}}{!r:16}\n'.format( '', el, l=lvl*8 ))
else:
out.append(' {}'.format(repr(el)))
out.append(f' {el!r}')
elif isList(el) or isDict(el):
indent = 1 if is_dict else lvl*8+4
out.append('{:>{l}}{:16}'.format('','<'+type(el).__name__+'>',l=indent))
out.append('{:>{l}}{:16}'.format( '', f'<{type(el).__name__}>', l=indent ))
if isList(el) and isinstance(el[0],scalars):
out.append('\n')
do_list(out,el,lvl=lvl+1,is_dict=isDict(el))
else:
out.append('{:>{l}}{:16} {}\n'.format(
'','<'+type(el).__name__+'>',repr(el),l=(lvl*8)+8))
out.append('{:>{l}}{:16} {!r}\n'.format( '', f'<{type(el).__name__}>', el, l=(lvl*8)+8 ))
out.append('\n')
if not e: out.append('{}\n'.format(repr(e)))
if not e:
out.append(f'{e!r}\n')
def isDict(obj):
return isinstance(obj,dict)
@ -90,7 +93,8 @@ if os.getenv('MMGEN_DEBUG') or os.getenv('MMGEN_TEST_SUITE') or os.getenv('MMGEN
def isScalar(obj):
return isinstance(obj,scalars)
out = ['<{}>{}\n'.format(type(self).__name__,' '+repr(self) if isScalar(self) else '')]
out = [f'<{type(self).__name__}>{" "+repr(self) if isScalar(self) else ""}\n']
if id(self) in id_list:
return out[-1].rstrip() + ' [RECURSION]\n'
if isList(self) or isDict(self):
@ -99,14 +103,21 @@ if os.getenv('MMGEN_DEBUG') or os.getenv('MMGEN_TEST_SUITE') or os.getenv('MMGEN
for k in self.__dict__:
e = getattr(self,k)
if isList(e) or isDict(e):
out.append('{:>{l}}{:<10} {:16}'.format('',k,'<'+type(e).__name__+'>',l=(lvl*8)+4))
out.append('{:>{l}}{:<10} {:16}'.format( '', k, f'<{type(e).__name__}>', l=(lvl*8)+4 ))
do_list(out,e,lvl=lvl,is_dict=isDict(e))
elif hasattr(e,'pfmt') and type(e) != type:
out.append('{:>{l}}{:10} {}'.format(
'',k,e.pfmt(lvl=lvl+1,id_list=id_list+[id(self)]),l=(lvl*8)+4))
'',
k,
e.pfmt( lvl=lvl+1, id_list=id_list+[id(self)] ),
l = (lvl*8)+4 ))
else:
out.append('{:>{l}}{:<10} {:16} {}\n'.format(
'',k,'<'+type(e).__name__+'>',repr(e),l=(lvl*8)+4))
'',
k,
f'<{type(e).__name__}>',
repr(e),
l=(lvl*8)+4 ))
import re
return re.sub('\n+','\n',''.join(out))
@ -123,11 +134,11 @@ if os.getenv('MMGEN_DEBUG') or os.getenv('MMGEN_TEST_SUITE') or os.getenv('MMGEN
attr = o.__dict__[attrname]
break
else:
rdie(3,'unable to find descriptor {}.{}'.format(cls.__name__,attrname))
rdie(3,f'unable to find descriptor {cls.__name__}.{attrname}')
if type(attr).__name__ == 'ImmutableAttr':
if attrname not in self.__dict__:
fs = 'attribute {!r} of {} has not been initialized in constructor!'
rdie(3,fs.format(attrname,cls.__name__))
rdie(3,
f'attribute {attrname!r} of {cls.__name__} has not been initialized in constructor!')
def print_diff(a,b,from_file='',to_file='',from_json=True):
if from_json:
@ -136,7 +147,8 @@ if os.getenv('MMGEN_DEBUG') or os.getenv('MMGEN_TEST_SUITE') or os.getenv('MMGEN
else:
a = a.split('\n')
b = b.split('\n')
sys.stderr.write(' DIFF:\n {}\n'.format('\n '.join(unified_diff(a,b,from_file,to_file))))
sys.stderr.write(' DIFF:\n {}\n'.format(
'\n '.join(unified_diff(a,b,from_file,to_file)) ))
def get_ndiff(a,b):
a = a.split('\n')

25
mmgen/filename.py
View file

@ -47,20 +47,19 @@ class Filename(MMGenObject):
self.ftype = ftype
# elif: # other MMGen file types
else:
die(3,"'{}': not a recognized file type for Wallet".format(ftype))
die(3,f'{ftype!r}: not a recognized file type for Wallet')
else:
die(3,"'{}': not a class".format(ftype))
die(3,f'{ftype!r}: not a class')
else:
# TODO: other file types
self.ftype = Wallet.ext_to_type(self.ext)
if not self.ftype:
m = "'{}': not a recognized Wallet file extension".format(self.ext)
raise BadFileExtension(m)
raise BadFileExtension(f'{self.ext!r}: not a recognized Wallet file extension')
try:
st = os.stat(fn)
except:
raise FileNotFound('{!r}: file not found'.format(fn))
raise FileNotFound(f'{fn!r}: file not found')
import stat
if stat.S_ISBLK(st.st_mode):
@ -70,7 +69,7 @@ class Filename(MMGenObject):
fd = os.open(fn, mode)
except OSError as e:
if e.errno == 13:
die(2,"'{}': permission denied".format(fn))
die(2,f'{fn!r}: permission denied')
# if e.errno != 17: raise
else:
self.size = os.lseek(fd, 0, os.SEEK_END)
@ -92,25 +91,27 @@ class MMGenFileList(list,MMGenObject):
def sort_by_age(self,key='mtime',reverse=False):
if key not in ('atime','ctime','mtime'):
die(1,"'{}': illegal sort key".format(key))
die(1,f'{key!r}: illegal sort key')
self.sort(key=lambda a: getattr(a,key),reverse=reverse)
def find_files_in_dir(ftype,fdir,no_dups=False):
if not isinstance(ftype,type):
die(3,"'{}': is of type {} (not a subclass of type 'type')".format(ftype,type(ftype)))
die(3,f"{ftype!r}: is of type {type(ftype)} (not a subclass of type 'type')")
from .wallet import Wallet
if not issubclass(ftype,Wallet):
die(3,"'{}': not a recognized file type".format(ftype))
die(3,f'{ftype!r}: not a recognized file type')
try: dirlist = os.listdir(fdir)
except: die(3,"ERROR: unable to read directory '{}'".format(fdir))
try:
dirlist = os.listdir(fdir)
except:
die(3,f'ERROR: unable to read directory {fdir!r}')
matches = [l for l in dirlist if l[-len(ftype.ext)-1:]=='.'+ftype.ext]
if no_dups:
if len(matches) > 1:
die(1,"ERROR: more than one {} file in directory '{}'".format(ftype.__name__,fdir))
die(1,f'ERROR: more than one {ftype.__name__} file in directory {fdir!r}')
return os.path.join(fdir,matches[0]) if len(matches) else None
else:
return [os.path.join(fdir,m) for m in matches]

6
mmgen/license.py
View file

@ -22,11 +22,11 @@ license.py: Copyright notice and text of GPLv3
from .globalvars import g
warning = """
{pnm} Copyright (C) {g.Cdates} by {g.author} {g.email}. This
warning = f"""
{g.proj_name} Copyright (C) {g.Cdates} by {g.author} {g.email}. This
program comes with ABSOLUTELY NO WARRANTY. This is free software, and
you are welcome to redistribute it under certain conditions.
""".format(g=g,pnm=g.proj_name)
"""
conditions = """
TERMS AND CONDITIONS

2
mmgen/main_addrgen.py
View file

@ -127,7 +127,7 @@ FMT CODES:
cmd_args = opts.init(opts_data,add_opts=['b16'],opt_filter=opt_filter)
errmsg = "'{}': invalid parameter for --type option".format(opt.type)
errmsg = f'{opt.type!r}: invalid parameter for --type option'
from .protocol import init_proto_from_opts
proto = init_proto_from_opts()

26
mmgen/main_addrimport.py
View file

@ -38,10 +38,10 @@ WARNING: If any of the addresses you're importing is already in the blockchain,
has a balance and is not in your tracking wallet, you must exit the program now
and rerun it using the '--rescan' option.
""".strip(),
'bad_args': """
You must specify an {pnm} address file, a single address with the '--address'
option, or a list of non-{pnm} addresses with the '--addrlist' option
""".strip().format(pnm=g.proj_name)
'bad_args': f"""
You must specify an {g.proj_name} address file, a single address with the '--address'
option, or a list of non-{g.proj_name} addresses with the '--addrlist' option
""".strip()
}[k]
# In batch mode, daemon just rescans each address separately anyway, so make
@ -49,7 +49,7 @@ option, or a list of non-{pnm} addresses with the '--addrlist' option
opts_data = {
'text': {
'desc': """Import addresses into an {} tracking wallet""".format(g.proj_name),
'desc': f'Import addresses into an {g.proj_name} tracking wallet',
'usage':'[opts] [mmgen address file]',
'options': """
-h, --help Print this help message
@ -87,7 +87,7 @@ def parse_cmd_args(rpc,cmd_args):
if opt.addrlist:
al = AddrList(
proto = proto,
addrlist = get_lines_from_file(infile,'non-{pnm} addresses'.format(pnm=g.proj_name),
addrlist = get_lines_from_file(infile,f'non-{g.proj_name} addresses',
trim_comments = True) )
else:
al = import_mmgen_list(infile)
@ -104,14 +104,14 @@ def check_opts(tw):
rescan = bool(opt.rescan)
if rescan and not 'rescan' in tw.caps:
msg("'--rescan' ignored: not supported by {}".format(type(tw).__name__))
msg(f"'--rescan' ignored: not supported by {type(tw).__name__}")
rescan = False
if rescan and not opt.quiet:
confirm_or_raise(ai_msgs('rescan'),'continue',expect='YES')
if batch and not 'batch' in tw.caps:
msg("'--batch' ignored: not supported by {}".format(type(tw).__name__))
msg(f"'--batch' ignored: not supported by {type(tw).__name__}")
batch = False
return batch,rescan
@ -124,7 +124,9 @@ async def import_addr(tw,addr,label,rescan,msg_fmt,msg_args):
while True:
if task.done():
break
msg_r(('\r{} '+msg_fmt).format(secs_to_hms(int(time.time()-start)),*msg_args))
msg_r(('\r{} '+msg_fmt).format(
secs_to_hms(int(time.time()-start)),
*msg_args ))
await asyncio.sleep(0.5)
await task
msg('\nOK')
@ -132,7 +134,7 @@ async def import_addr(tw,addr,label,rescan,msg_fmt,msg_args):
await task
qmsg(msg_fmt.format(*msg_args) + ' - OK')
except Exception as e:
die(2,'\nImport of address {!r} failed: {!r}'.format(addr,e.args[0]))
die(2,f'\nImport of address {addr!r} failed: {e.args[0]!r}')
def make_args_list(tw,al,batch,rescan):
@ -142,10 +144,10 @@ def make_args_list(tw,al,batch,rescan):
for num,e in enumerate(al.data,1):
if e.idx:
label = '{}:{}'.format(al.al_id,e.idx) + (' ' + e.label if e.label else '')
label = f'{al.al_id}:{e.idx}' + (' ' + e.label if e.label else '')
add_msg = label
else:
label = '{}:{}'.format(proto.base_coin.lower(),e.addr)
label = f'{proto.base_coin.lower()}:{e.addr}'
add_msg = 'non-'+g.proj_name
if batch:

24
mmgen/main_autosign.py
View file

@ -44,15 +44,15 @@ opts_data = {
'text': {
'desc': 'Auto-sign MMGen transactions',
'usage':'[opts] [command]',
'options': """
'options': f"""
-h, --help Print this help message
--, --longhelp Print help message for long options (common options)
-c, --coins=c Coins to sign for (comma-separated list)
-I, --no-insert-check Don’t check for device insertion
-l, --led Use status LED to signal standby, busy and error
-m, --mountpoint=M Specify an alternate mountpoint 'M' (default: '{mp}')
-m, --mountpoint=M Specify an alternate mountpoint 'M' (default: '{mountpoint}')
-M, --mnemonic-fmt=F During setup, prompt for mnemonic seed phrase of format
'F' (choices: {mc}; default: '{md}')
'F' (choices: {fmt_list(mn_fmts,fmt='no_spc')}; default: {mn_fmt_dfl!r})
-n, --no-summary Don’t print a transaction summary
-s, --stealth-led Stealth LED mode - signal busy and error only, and only
after successful authorization.
@ -61,16 +61,12 @@ opts_data = {
will not be printed.
-q, --quiet Produce quieter output
-v, --verbose Produce more verbose output
""".format(
md = mn_fmt_dfl,
mc = fmt_list(mn_fmts,fmt='no_spc'),
mp = mountpoint
),
'notes': """
""",
'notes': f"""
COMMANDS
gen_key - generate the wallet encryption key and copy it to '{td}'
gen_key - generate the wallet encryption key and copy it to '{tx_dir}'
setup - generate the wallet encryption key and wallet
wait - start in loop mode: wait-mount-sign-unmount-wait
@ -91,13 +87,13 @@ ready for device insertion or removal.
The removable device must have a partition labeled MMGEN_TX and a user-
writable directory '/tx', where unsigned MMGen transactions are placed.
On the signing machine the mount point '{mp}' must exist and /etc/fstab
On the signing machine the mount point '{mountpoint}' must exist and /etc/fstab
must contain the following entry:
LABEL='MMGEN_TX' /mnt/tx auto noauto,user 0 0
Transactions are signed with a wallet on the signing machine (in the directory
'{wd}') encrypted with a 64-character hexadecimal password on the
'{wallet_dir}') encrypted with a 64-character hexadecimal password on the
removable device.
The password and wallet can be created in one operation by invoking the
@ -108,7 +104,7 @@ Alternatively, the password and wallet can be created separately by first
invoking the command with 'gen_key' and then creating and encrypting the
wallet using the -P (--passwd-file) option:
$ mmgen-walletconv -r0 -q -iwords -d{wd} -p1 -P{td}/{kf} -Llabel
$ mmgen-walletconv -r0 -q -iwords -d{wallet_dir} -p1 -P{tx_dir}/{key_fn} -Llabel
Note that the hash preset must be '1'. Multiple wallets are permissible.
@ -116,7 +112,7 @@ For good security, it's advisable to re-generate a new wallet and key for
each signing session.
This command is currently available only on Linux-based platforms.
""".format(pnm=prog_name,wd=wallet_dir,td=tx_dir,kf=key_fn,mp=mountpoint)
"""
}
}

8
mmgen/main_passgen.py
View file

@ -33,10 +33,10 @@ pwi_fs = '{:8} {:1} {:26} {:<7} {:<7} {}'
opts_data = {
'sets': [('print_checksum',True,'quiet',True)],
'text': {
'desc': """
Generate a range or list of passwords from an {pnm} wallet,
'desc': f"""
Generate a range or list of passwords from an {g.proj_name} wallet,
mnemonic, seed or brainwallet for the given ID string
""".format(pnm=g.proj_name),
""",
'usage':'[opts] [seed source] <ID string> <index list or range(s)>',
'options': """
-h, --help Print this help message
@ -118,7 +118,7 @@ FMT CODES:
seed_lens=', '.join(map(str,g.seed_lens)),
g=g,pnm=g.proj_name,
dpf=PasswordList.dfl_pw_fmt,
kgs=' '.join(['{}:{}'.format(n,k) for n,k in enumerate(g.key_generators,1)])
kgs=' '.join([f'{n}:{k}' for n,k in enumerate(g.key_generators,1)])
),
'notes': lambda help_notes,s: s.format(
o=opts,g=g,i58=pwi['b58'],i32=pwi['b32'],i39=pwi['bip39'],

4
mmgen/main_regtest.py
View file

@ -26,7 +26,7 @@ from .common import *
opts_data = {
'sets': [('yes', True, 'quiet', True)],
'text': {
'desc': 'Coin daemon regression test mode setup and operations for the {} suite'.format(g.proj_name),
'desc': f'Coin daemon regression test mode setup and operations for the {g.proj_name} suite',
'usage': '[opts] <command>',
'options': """
-h, --help Print this help message
@ -77,7 +77,7 @@ def check_num_args():
if not cmd_args:
opts.usage()
elif cmd_args[0] not in MMGenRegtest.usr_cmds:
die(1,'{!r}: invalid command'.format(cmd_args[0]))
die(1,f'{cmd_args[0]!r}: invalid command')
elif cmd_args[0] not in ('cli','balances'):
check_num_args()

4
mmgen/main_seedjoin.py
View file

@ -102,7 +102,7 @@ def print_shares_info():
len(shares) )
si = 1
for n,s in enumerate(shares[si:],si+1):
out += '{:3}: {}\n'.format(n,s.sid)
out += f'{n:3}: {s.sid}\n'
qmsg(out)
cmd_args = opts.init(opts_data)
@ -139,6 +139,6 @@ msg_r('Joining {n}-of-{n} XOR split...'.format(n=len(shares)))
seed_out = Seed.join_shares([share1]+shares[1:])
msg('OK\nJoined Seed ID: {}'.format(seed_out.sid.hl()))
msg(f'OK\nJoined Seed ID: {seed_out.sid.hl()}')
Wallet(seed=seed_out).write_to_file()

23
mmgen/main_split.py
View file

@ -29,10 +29,10 @@ from .common import *
opts_data = {
'text': {
'desc': """
Split funds in a {pnm} wallet after a chain fork using a
'desc': f"""
Split funds in an {g.proj_name} wallet after a chain fork using a
timelocked transaction
""".format(pnm=g.proj_name),
""",
'usage':'[opts] [output addr1] [output addr2]',
'options': """
-h, --help Print this help message
@ -51,10 +51,10 @@ opts_data = {
-L, --locktime= t Lock time (block height or unix seconds)
(default: {bh})
""",
'notes': """\n
'notes': f"""\n
This command creates two transactions: one (with the timelock) to be broadcast
on the long chain and one on the short chain after a replayable chain fork.
Only {pnm} addresses may be spent to.
Only {g.proj_name} addresses may be spent to.
The command must be run on the longest chain. The user is reponsible for
ensuring that the current chain is the longest. The other chain is specified
@ -77,7 +77,7 @@ attacks on the majority chain or reorg attacks on the minority chain if the
minority chain is ahead of the timelock. If the reorg'd minority chain is
behind the timelock, protection is contingent on getting the non-timelocked
transaction reconfirmed before the timelock expires. Use at your own risk.
""".format(pnm=g.proj_name)
"""
},
'code': {
'options': lambda proto,s: s.format(
@ -96,11 +96,10 @@ die(1,'This command is disabled')
# the following code is broken:
opt.other_coin = opt.other_coin.upper() if opt.other_coin else proto.forks[-1][2].upper()
if opt.other_coin.lower() not in [e[2] for e in proto.forks if e[3] == True]:
die(1,"'{}': not a replayable fork of {} chain".format(opt.other_coin,proto.coin))
die(1,f'{opt.other_coin!r}: not a replayable fork of {proto.coin} chain')
if len(cmd_args) != 2:
fs = 'This command requires exactly two {} addresses as arguments'
die(1,fs.format(g.proj_name))
die(1,f'This command requires exactly two {g.proj_name} addresses as arguments')
from .obj import MMGenID
try:
@ -109,7 +108,7 @@ except:
die(1,'Command line arguments must be valid MMGen IDs')
if mmids[0] == mmids[1]:
die(2,'Both transactions have the same output! ({})'.format(mmids[0]))
die(2,f'Both transactions have the same output! ({mmids[0]})')
from .tx import MMGenSplitTX
from .protocol import init_proto
@ -124,7 +123,7 @@ tx1 = MMGenSplitTX()
opt.no_blank = True
async def main():
gmsg("Creating timelocked transaction for long chain ({})".format(proto.coin))
gmsg(f'Creating timelocked transaction for long chain ({proto.coin})')
locktime = int(opt.locktime)
if not locktime:
rpc = rpc_init(proto)
@ -134,7 +133,7 @@ async def main():
tx1.format()
tx1.create_fn()
gmsg("\nCreating transaction for short chain ({})".format(opt.other_coin))
gmsg(f'\nCreating transaction for short chain ({opt.other_coin})')
proto = init_proto(opt.other_coin)

17
mmgen/main_tool.py
View file

@ -25,7 +25,7 @@ from .common import *
def make_cmd_help():
import mmgen.tool
def make_help():
def do():
for bc in mmgen.tool.MMGenToolCmds.classes.values():
cls_doc = bc.__doc__.strip().split('\n')
for l in cls_doc:
@ -40,22 +40,23 @@ def make_cmd_help():
yield ''
max_w = max(map(len,bc.user_commands))
fs = ' {{:{}}} - {{}}'.format(max_w)
for name,code in sorted(bc.user_commands.items()):
if code.__doc__:
yield fs.format(name,
yield ' {:{}} - {}'.format(
name,
max_w,
pretty_format(
code.__doc__.strip().replace('\n\t\t',' '),
width=79-(max_w+7),
pfx=' '*(max_w+5)).lstrip()
width = 79-(max_w+7),
pfx = ' '*(max_w+5)).lstrip()
)
yield ''
return '\n'.join(make_help())
return '\n'.join(do())
opts_data = {
'text': {
'desc': 'Perform various {pnm}- and cryptocoin-related operations'.format(pnm=g.proj_name),
'desc': f'Perform various {g.proj_name}- and cryptocoin-related operations',
'usage': '[opts] <command> <command args>',
'options': """
-d, --outdir= d Specify an alternate directory 'd' for output
@ -103,7 +104,7 @@ if cmd in ('help','usage') and cmd_args:
cmd_args[0] = 'command_name=' + cmd_args[0]
if cmd not in tool.MMGenToolCmds:
die(1,"'{}': no such command".format(cmd))
die(1,f'{cmd!r}: no such command')
args,kwargs = tool._process_args(cmd,cmd_args)

10
mmgen/main_txbump.py
View file

@ -27,12 +27,12 @@ from .wallet import Wallet
opts_data = {
'sets': [('yes', True, 'quiet', True)],
'text': {
'desc': """
'desc': f"""
Increase the fee on a replaceable (RBF) {g.proj_name} transaction,
creating a new transaction, and optionally sign and send the
new transaction
""".format(g=g),
'usage': '[opts] <{g.proj_name} TX file> [seed source] ...'.format(g=g),
""",
'usage': f'[opts] <{g.proj_name} TX file> [seed source] ...',
'options': """
-h, --help Print this help message
--, --longhelp Print help message for long options (common options)
@ -87,7 +87,7 @@ column below:
pnl=g.proj_name.lower(),
fu=help_notes('rel_fee_desc'),
fl=help_notes('fee_spec_letters'),
kgs=' '.join(['{}:{}'.format(n,k) for n,k in enumerate(g.key_generators,1)]),
kgs=' '.join([f'{n}:{k}' for n,k in enumerate(g.key_generators,1)]),
kg=g.key_generator,
cu=proto.coin),
'notes': lambda help_notes,s: s.format(
@ -147,7 +147,7 @@ async def main():
output_idx = tx.choose_output()
if not silent:
msg('Minimum fee for new transaction: {} {}'.format(tx.min_fee.hl(),tx.proto.coin))
msg(f'Minimum fee for new transaction: {tx.min_fee.hl()} {tx.proto.coin}')
tx.usr_fee = tx.get_usr_fee_interactive(tx_fee=opt.tx_fee,desc='User-selected')

2
mmgen/main_txcreate.py
View file

@ -26,7 +26,7 @@ from .common import *
opts_data = {
'sets': [('yes', True, 'quiet', True)],
'text': {
'desc': 'Create a transaction with outputs to specified coin or {g.proj_name} addresses'.format(g=g),
'desc': f'Create a transaction with outputs to specified coin or {g.proj_name} addresses',
'usage': '[opts] <addr,amt> ... [change addr] [addr file] ...',
'options': """
-h, --help Print this help message

4
mmgen/main_txdo.py
View file

@ -27,7 +27,7 @@ from .obj import SubSeedIdxRange
opts_data = {
'sets': [('yes', True, 'quiet', True)],
'text': {
'desc': 'Create, sign and send an {g.proj_name} transaction'.format(g=g),
'desc': f'Create, sign and send an {g.proj_name} transaction',
'usage': '[opts] <addr,amt> ... [change addr] [addr file] ... [seed source] ...',
'options': """
-h, --help Print this help message
@ -95,7 +95,7 @@ column below:
'code': {
'options': lambda proto,help_notes,s: s.format(
g=g,pnm=g.proj_name,pnl=g.proj_name.lower(),
kgs=' '.join(['{}:{}'.format(n,k) for n,k in enumerate(g.key_generators,1)]),
kgs=' '.join([f'{n}:{k}' for n,k in enumerate(g.key_generators,1)]),
fu=help_notes('rel_fee_desc'),
fl=help_notes('fee_spec_letters'),
ss=g.subseeds,

2
mmgen/main_txsend.py
View file

@ -25,7 +25,7 @@ from .common import *
opts_data = {
'sets': [('yes', True, 'quiet', True)],
'text': {
'desc': 'Send a signed {pnm} cryptocoin transaction'.format(pnm=g.proj_name),
'desc': f'Send a signed {g.proj_name} cryptocoin transaction',
'usage': '[opts] <signed transaction file>',
'options': """
-h, --help Print this help message

4
mmgen/main_txsign.py
View file

@ -28,7 +28,7 @@ from .wallet import Wallet
opts_data = {
'sets': [('yes', True, 'quiet', True)],
'text': {
'desc': 'Sign cryptocoin transactions generated by {pnl}-txcreate'.format(pnl=g.proj_name.lower()),
'desc': f'Sign cryptocoin transactions generated by {g.proj_name.lower()}-txcreate',
'usage': '[opts] <transaction file>... [seed source]...',
'options': """
-h, --help Print this help message
@ -81,7 +81,7 @@ column below:
g=g,
pnm=g.proj_name,
pnl=g.proj_name.lower(),
kgs=' '.join(['{}:{}'.format(n,k) for n,k in enumerate(g.key_generators,1)]),
kgs=' '.join([f'{n}:{k}' for n,k in enumerate(g.key_generators,1)]),
kg=g.key_generator,
ss=g.subseeds,
ss_max=SubSeedIdxRange.max_idx,

8
mmgen/main_wallet.py
View file

@ -43,11 +43,11 @@ invoked_as = {
'mmgen-seedsplit': 'seedsplit',
}[g.prog_name]
dsw = 'the default or specified {pnm} wallet'
dsw = f'the default or specified {g.proj_name} wallet'
# full: defhHiJkKlLmoOpPqrSvz-
if invoked_as == 'gen':
desc = 'Generate an {pnm} wallet from a random seed'
desc = f'Generate an {g.proj_name} wallet from a random seed'
opt_filter = 'ehdoJlLpPqrSvz-'
usage = '[opts]'
oaction = 'output'
@ -81,7 +81,7 @@ elif invoked_as == 'seedsplit':
opts_data = {
'text': {
'desc': desc.format(pnm=g.proj_name),
'desc': desc,
'usage': usage,
'options': """
-h, --help Print this help message
@ -187,7 +187,7 @@ else:
if invoked_as == 'chk':
lbl = ss_in.ssdata.label.hl() if hasattr(ss_in.ssdata,'label') else 'NONE'
vmsg('Wallet label: {}'.format(lbl))
vmsg(f'Wallet label: {lbl}')
# TODO: display creation date
sys.exit(0)

36
mmgen/mn_entry.py
View file

@ -219,7 +219,7 @@ def mn_entry(wl_id,entry_mode=None):
wl_id = 'mmgen'
me = MnemonicEntry.get_cls_by_wordlist(wl_id)
import importlib
me.conv_cls = getattr(importlib.import_module('mmgen.{}'.format(me.modname)),me.modname)
me.conv_cls = getattr(importlib.import_module(f'mmgen.{me.modname}'),me.modname)
me.conv_cls.init_mn(wl_id)
me.wl = me.conv_cls.digits[wl_id]
obj = me()
@ -336,7 +336,7 @@ class MnemonicEntry(object):
def get_mnemonic_from_user(self,mn_len,validate=True):
mll = list(self.conv_cls.seedlen_map_rev[self.wl_id])
assert mn_len in mll, '{}: invalid mnemonic length (must be one of {})'.format(mn_len,mll)
assert mn_len in mll, f'{mn_len}: invalid mnemonic length (must be one of {mll})'
if self.usr_dfl_entry_mode:
em = self.get_cls_by_entry_mode(self.usr_dfl_entry_mode)(self)
@ -345,20 +345,19 @@ class MnemonicEntry(object):
em = self.choose_entry_mode()
i_add = '.'
msg('\r' + 'Using {} entry mode{}'.format(cyan(em.name.upper()),i_add))
msg('\r' + f'Using {cyan(em.name.upper())} entry mode{i_add}')
self.em = em
if not self.usr_dfl_entry_mode:
m = (
fmt(self.prompt_info['intro'])
+ '\n'
+ fmt(self.prompt_info['pad_info'].rstrip() + em.pad_max_info + em.prompt_info, indent=' ')
)
msg('\n' + m.format(
ml = mn_len,
ssl = em.ss_len,
pad_max = em.pad_max,
sw = self.shortest_word,
msg('\n' + (
fmt(self.prompt_info['intro'])
+ '\n'
+ fmt(self.prompt_info['pad_info'].rstrip() + em.pad_max_info + em.prompt_info, indent=' ')
).format(
ml = mn_len,
ssl = em.ss_len,
pad_max = em.pad_max,
sw = self.shortest_word,
))
clear_line = '\n' if g.test_suite else '{r}{s}{r}'.format(r='\r',s=' '*40)
@ -393,8 +392,7 @@ class MnemonicEntry(object):
}
wl = wl.lower()
if wl not in d:
m = 'wordlist {!r} not recognized (valid options: {})'
raise ValueError(m.format(wl,fmt_list(list(d))))
raise ValueError(f'wordlist {wl!r} not recognized (valid options: {fmt_list(list(d))})')
return d[wl]
@classmethod
@ -402,8 +400,9 @@ class MnemonicEntry(object):
for k,v in g.mnemonic_entry_modes.items():
tcls = cls.get_cls_by_wordlist(k)
if v not in tcls.entry_modes:
m = 'entry mode {!r} not recognized for wordlist {!r}:\n (valid options: {})'
raise ValueError(m.format(v,k,fmt_list(tcls.entry_modes)))
raise ValueError(
f'entry mode {v!r} not recognized for wordlist {k!r}:' +
f'\n (valid options: {fmt_list(tcls.entry_modes)})' )
tcls.usr_dfl_entry_mode = v
class MnemonicEntryMMGen(MnemonicEntry):
@ -430,5 +429,4 @@ class MnemonicEntryMonero(MnemonicEntry):
try:
MnemonicEntry.get_cfg_vars()
except Exception as e:
m = "Error in cfg file option 'mnemonic_entry_modes':\n {}"
die(2,m.format(e.args[0]))
die(2, f"Error in cfg file option 'mnemonic_entry_modes':\n {e.args[0]}")

6
mmgen/obj.py
View file

@ -494,7 +494,7 @@ class CoinAmt(Decimal,Hilite,InitErrors): # abstract class
def to_unit(self,unit,show_decimal=False):
ret = Decimal(self) // getattr(self,unit)
if show_decimal and ret < 1:
return '{:.8f}'.format(ret).rstrip('0')
return f'{ret:.8f}'.rstrip('0')
return int(ret)
@classmethod
@ -675,8 +675,8 @@ class SubSeedIdx(str,Hilite,InitErrors):
from .util import is_int
assert is_int(idx),"valid format: an integer, plus optional letter 'S','s','L' or 'l'"
idx = int(idx)
assert idx >= SubSeedIdxRange.min_idx, 'subseed index < {:,}'.format(SubSeedIdxRange.min_idx)
assert idx <= SubSeedIdxRange.max_idx, 'subseed index > {:,}'.format(SubSeedIdxRange.max_idx)
assert idx >= SubSeedIdxRange.min_idx, f'subseed index < {SubSeedIdxRange.min_idx:,}'
assert idx <= SubSeedIdxRange.max_idx, f'subseed index > {SubSeedIdxRange.max_idx:,}'
sstype,ltr = ('short','S') if s[-1] in 'Ss' else ('long','L')
me = str.__new__(cls,str(idx)+ltr)

88
mmgen/opts.py
View file

@ -40,11 +40,11 @@ def usage():
Die(1,Opts.make_usage_str(g.prog_name,'user',usage_data))
def version():
Die(0,fmt("""
{pn} version {g.version}
Die(0,fmt(f"""
{g.prog_name.upper()} version {g.version}
Part of the {g.proj_name} suite, an online/offline cryptocurrency wallet for the
command line. Copyright (C){g.Cdates} {g.author} {g.email}
""".format(g=g,pn=g.prog_name.upper()),indent=' ').rstrip())
""",indent=' ').rstrip())
def print_help(po,opts_data,opt_filter):
if not 'code' in opts_data:
@ -167,14 +167,9 @@ def show_common_opts_diff():
a = set(g.common_opts)
b = set(common_opts_data_to_list())
m1 = 'g.common_opts - common_opts_data:\n {}\n'
msg(m1.format(do_fmt(a-b) if a-b else 'None'))
m2 = 'common_opts_data - g.common_opts (these do not set global var):\n{}\n'
msg(m2.format(do_fmt(b-a)))
m3 = 'common_opts_data ^ g.common_opts (these set global var):\n{}\n'
msg(m3.format(do_fmt(b.intersection(a))))
msg(f'g.common_opts - common_opts_data:\n {do_fmt(a-b) if a-b else "None"}\n')
msg(f'common_opts_data - g.common_opts (these do not set global var):\n{do_fmt(b-a)}\n')
msg(f'common_opts_data ^ g.common_opts (these set global var):\n{do_fmt(b.intersection(a))}\n')
sys.exit(0)
@ -407,11 +402,18 @@ def opt_is_tx_fee(key,val,desc): # 'key' must remain a placeholder
if ret == False:
raise UserOptError('{!r}: invalid {}\n(not a {} amount or {} specification)'.format(
val,desc,tx.proto.coin.upper(),tx.rel_fee_desc))
val,
desc,
tx.proto.coin.upper(),
tx.rel_fee_desc ))
if ret > tx.proto.max_tx_fee:
raise UserOptError('{!r}: invalid {}\n({} > max_tx_fee ({} {}))'.format(
val,desc,ret.fmt(fs='1.1'),tx.proto.max_tx_fee,tx.proto.coin.upper()))
val,
desc,
ret.fmt(fs='1.1'),
tx.proto.max_tx_fee,
tx.proto.coin.upper() ))
def check_usr_opts(usr_opts): # Raises an exception if any check fails
@ -420,40 +422,45 @@ def check_usr_opts(usr_opts): # Raises an exception if any check fails
try:
l = val.split(sep)
except:
raise UserOptError('{!r}: invalid {} (not {}-separated list)'.format(val,desc,sepword))
raise UserOptError(f'{val!r}: invalid {desc} (not {sepword}-separated list)')
if len(l) != n:
raise UserOptError('{!r}: invalid {} ({} {}-separated items required)'.format(val,desc,n,sepword))
raise UserOptError(f'{val!r}: invalid {desc} ({n} {sepword}-separated items required)')
def opt_compares(val,op_str,target,desc,desc2=''):
import operator as o
op_f = { '<':o.lt, '<=':o.le, '>':o.gt, '>=':o.ge, '=':o.eq }[op_str]
if not op_f(val,target):
d2 = desc2 + ' ' if desc2 else ''
raise UserOptError('{}: invalid {} ({}not {} {})'.format(val,desc,d2,op_str,target))
raise UserOptError(f'{val}: invalid {desc} ({d2}not {op_str} {target})')
def opt_is_int(val,desc):
if not is_int(val):
raise UserOptError('{!r}: invalid {} (not an integer)'.format(val,desc))
raise UserOptError(f'{val!r}: invalid {desc} (not an integer)')
def opt_is_float(val,desc):
try:
float(val)
except:
raise UserOptError('{!r}: invalid {} (not a floating-point number)'.format(val,desc))
raise UserOptError(f'{val!r}: invalid {desc} (not a floating-point number)')
def opt_is_in_list(val,tlist,desc):
if val not in tlist:
q,sep = (('',','),("'","','"))[type(tlist[0]) == str]
fs = '{q}{v}{q}: invalid {w}\nValid choices: {q}{o}{q}'
raise UserOptError(fs.format(v=val,w=desc,q=q,o=sep.join(map(str,sorted(tlist)))))
raise UserOptError('{q}{v}{q}: invalid {w}\nValid choices: {q}{o}{q}'.format(
v = val,
w = desc,
q = q,
o = sep.join(map(str,sorted(tlist))) ))
def opt_unrecognized(key,val,desc='value'):
raise UserOptError('{!r}: unrecognized {} for option {!r}'.format(val,desc,fmt_opt(key)))
raise UserOptError(f'{val!r}: unrecognized {desc} for option {fmt_opt(key)!r}')
def opt_display(key,val='',beg='For selected',end=':\n'):
s = '{}={}'.format(fmt_opt(key),val) if val else fmt_opt(key)
msg_r('{} option {!r}{}'.format(beg,s,end))
msg_r('{} option {!r}{}'.format(
beg,
f'{fmt_opt(key)}={val}' if val else fmt_opt(key),
end ))
def chk_in_fmt(key,val,desc):
from .wallet import Wallet,IncogWallet,Brainwallet,IncogWalletHidden
@ -463,9 +470,9 @@ def check_usr_opts(usr_opts): # Raises an exception if any check fails
if key == 'out_fmt':
p = 'hidden_incog_output_params'
if sstype == IncogWalletHidden and not getattr(opt,p):
m1 = 'Hidden incog format output requested. '
m2 = 'You must supply a file and offset with the {!r} option'
raise UserOptError(m1+m2.format(fmt_opt(p)))
raise UserOptError(
'Hidden incog format output requested. ' +
f'You must supply a file and offset with the {fmt_opt(p)!r} option' )
if issubclass(sstype,IncogWallet) and opt.old_incog_fmt:
opt_display(key,val,beg='Selected',end=' ')
opt_display('old_incog_fmt',beg='conflicts with',end=':\n')
@ -500,8 +507,7 @@ def check_usr_opts(usr_opts): # Raises an exception if any check fails
val2 = getattr(opt,key2)
from .wallet import IncogWalletHidden
if val2 and val2 not in IncogWalletHidden.fmt_codes:
fs = 'Option conflict:\n {}, with\n {}={}'
raise UserOptError(fs.format(fmt_opt(key),fmt_opt(key2),val2))
raise UserOptError(f'Option conflict:\n {fmt_opt(key)}, with\n {fmt_opt(key2)}={val2}')
chk_hidden_incog_output_params = chk_hidden_incog_input_params
@ -538,7 +544,7 @@ def check_usr_opts(usr_opts): # Raises an exception if any check fails
def chk_vsize_adj(key,val,desc):
opt_is_float(val,desc)
ymsg('Adjusting transaction vsize by a factor of {:1.2f}'.format(float(val)))
ymsg(f'Adjusting transaction vsize by a factor of {float(val):1.2f}')
def chk_key_generator(key,val,desc):
opt_compares(val,'<=',len(g.key_generators),desc)
@ -551,16 +557,16 @@ def check_usr_opts(usr_opts): # Raises an exception if any check fails
# TODO: move this check elsewhere
# def chk_rbf(key,val,desc):
# if not proto.cap('rbf'):
# m = '--rbf requested, but {} does not support replace-by-fee transactions'
# raise UserOptError(m.format(proto.coin))
# raise UserOptError(f'--rbf requested, but {proto.coin} does not support replace-by-fee transactions')
# def chk_bob(key,val,desc):
# m = "Regtest (Bob and Alice) mode not set up yet. Run '{}-regtest setup' to initialize."
# from .regtest import MMGenRegtest
# try:
# os.stat(os.path.join(MMGenRegtest(g.coin).d.datadir,'regtest','debug.log'))
# except:
# raise UserOptError(m.format(g.proj_name.lower()))
# raise UserOptError(
# 'Regtest (Bob and Alice) mode not set up yet. ' +
# f"Run '{g.proj_name.lower()}-regtest setup' to initialize." )
#
# chk_alice = chk_bob
@ -571,17 +577,17 @@ def check_usr_opts(usr_opts): # Raises an exception if any check fails
# TODO: move this check elsewhere
# def chk_token(key,val,desc):
# if not 'token' in proto.caps:
# raise UserOptError('Coin {!r} does not support the --token option'.format(tx.coin))
# raise UserOptError(f'Coin {tx.coin!r} does not support the --token option')
# if len(val) == 40 and is_hex_str(val):
# return
# if len(val) > 20 or not all(s.isalnum() for s in val):
# raise UserOptError('{!r}: invalid parameter for --token option'.format(val))
# raise UserOptError(f'{val!r}: invalid parameter for --token option')
cfuncs = { k:v for k,v in locals().items() if k.startswith('chk_') }
for key in usr_opts:
val = getattr(opt,key)
desc = 'parameter for {!r} option'.format(fmt_opt(key))
desc = f'parameter for {fmt_opt(key)!r} option'
if key in g.infile_opts:
check_infile(val) # file exists and is readable - dies on error
@ -590,7 +596,7 @@ def check_usr_opts(usr_opts): # Raises an exception if any check fails
elif 'chk_'+key in cfuncs:
cfuncs['chk_'+key](key,val,desc)
elif g.debug:
Msg('check_usr_opts(): No test for opt {!r}'.format(key))
Msg(f'check_usr_opts(): No test for opt {key!r}')
def set_auto_typeset_opts():
for key,ref_type in g.auto_typeset_opts.items():
@ -622,8 +628,12 @@ def check_and_set_autoset_opts(): # Raises exception if any check fails
else:
ret = locals()[asd.type](key,val,asd)
if type(ret) is str:
m = '{!r}: invalid parameter for option --{} (not {}: {})'
raise UserOptError(m.format(val,key.replace('_','-'),ret,fmt_list(asd.choices)))
raise UserOptError(
'{!r}: invalid parameter for option --{} (not {}: {})'.format(
val,
key.replace('_','-'),
ret,
fmt_list(asd.choices) ))
elif ret is True:
setattr(opt,key,val)
else:

22
mmgen/regtest.py
View file

@ -30,8 +30,8 @@ def create_data_dir(data_dir):
try: os.stat(os.path.join(data_dir,'regtest'))
except: pass
else:
m = "Delete your existing MMGen regtest setup at '{}' and create a new one?"
if keypress_confirm(m.format(data_dir)):
if keypress_confirm(
f'Delete your existing MMGen regtest setup at {data_dir!r} and create a new one?'):
shutil.rmtree(data_dir)
else:
die()
@ -81,7 +81,7 @@ class MMGenRegtest(MMGenObject):
if len(out) != blocks:
rdie(1,'Error generating blocks')
gmsg('Mined {} block{}'.format(blocks,suf(blocks)))
gmsg(f'Mined {blocks} block{suf(blocks)}')
async def setup(self):
@ -93,7 +93,7 @@ class MMGenRegtest(MMGenObject):
create_data_dir(self.d.datadir)
gmsg('Starting {} regtest setup'.format(self.coin.upper()))
gmsg(f'Starting {self.coin.upper()} regtest setup')
self.d.start(silent=True)
@ -164,7 +164,7 @@ class MMGenRegtest(MMGenObject):
msg(fs.format('Total balance:',sum(v for k,v in bal.items())))
async def send(self,addr,amt):
gmsg('Sending {} miner {} to address {}'.format(amt,self.d.coin,addr))
gmsg(f'Sending {amt} miner {self.d.coin} to address {addr}')
cp = await self.rpc_call('sendtoaddress',addr,str(amt),wallet='miner')
await self.generate(1)
@ -185,14 +185,16 @@ class MMGenRegtest(MMGenObject):
proto = init_proto(coin,False)
if not [f for f in proto.forks if f[2] == proto.coin.lower() and f[3] == True]:
die(1,"Coin {} is not a replayable fork of coin {}".format(proto.coin,coin))
die(1,f'Coin {proto.coin} is not a replayable fork of coin {coin}')
gmsg('Creating fork from coin {} to coin {}'.format(coin,proto.coin))
gmsg(f'Creating fork from coin {coin} to coin {proto.coin}')
source_rt = MMGenRegtest(coin)
try: os.stat(source_rt.d.datadir)
except: die(1,"Source directory '{}' does not exist!".format(source_rt.d.datadir))
try:
os.stat(source_rt.d.datadir)
except:
die(1,f'Source directory {source_rt.d.datadir!r} does not exist!')
# stop the source daemon
if source_rt.d.state != 'stopped':
@ -211,4 +213,4 @@ class MMGenRegtest(MMGenObject):
await self.start_daemon(reindex=True)
await self.rpc_call('stop')
gmsg('Fork {} successfully created'.format(proto.coin))
gmsg(f'Fork {proto.coin} successfully created')

13
mmgen/rpc.py
View file

@ -51,7 +51,10 @@ rpc_credentials_msg = '\n'+fmt("""
def dmsg_rpc(fs,data=None,is_json=False):
if g.debug_rpc:
msg(fs if data == None else fs.format(pp_fmt(json.loads(data) if is_json else data)))
msg(
fs if data == None else
fs.format(pp_fmt(json.loads(data) if is_json else data))
)
class json_encoder(json.JSONEncoder):
def default(self,obj):
@ -130,8 +133,10 @@ class RPCBackends:
auth_str = f'{caller.auth.user}:{caller.auth.passwd}'
auth_str_b64 = 'Basic ' + base64.b64encode(auth_str.encode()).decode()
self.http_hdrs.update({ 'Host': self.host, 'Authorization': auth_str_b64 })
fs = ' RPC AUTHORIZATION data ==> raw: [{}]\n{:>31}enc: [{}]\n'
dmsg_rpc(fs.format(auth_str,'',auth_str_b64))
dmsg_rpc(' RPC AUTHORIZATION data ==> raw: [{}]\n{:>31}enc: [{}]\n'.format(
auth_str,
'',
auth_str_b64 ))
async def run(self,payload,timeout,wallet):
dmsg_rpc('\n RPC PAYLOAD data (httplib) ==>\n{}\n',payload)
@ -247,7 +252,7 @@ class RPCClient(MMGenObject):
def __init__(self,host,port,test_connection=True):
dmsg_rpc('=== {}.__init__() debug ==='.format(type(self).__name__))
dmsg_rpc(f'=== {type(self).__name__}.__init__() debug ===')
dmsg_rpc(f' cls [{type(self).__name__}] host [{host}] port [{port}]\n')
if test_connection:

54
mmgen/seed.py
View file

@ -34,7 +34,7 @@ class SeedBase(MMGenObject):
# Truncate random data for smaller seed lengths
seed_bin = sha256(get_random(1033)).digest()[:(opt.seed_len or g.dfl_seed_len)//8]
elif len(seed_bin)*8 not in g.seed_lens:
die(3,'{}: invalid seed length'.format(len(seed_bin)))
die(3,f'{len(seed_bin)}: invalid seed length')
self.data = seed_bin
self.sid = SeedID(seed=self)
@ -83,14 +83,16 @@ class SubSeedList(MMGenObject):
sid = self.data[ss_idx.type].key(ss_idx.idx-1)
idx,nonce = self.data[ss_idx.type][sid]
if idx != ss_idx.idx:
m = "{} != {}: self.data[{t!r}].key(i) does not match self.data[{t!r}][i]!"
die(3,m.format(idx,ss_idx.idx,t=ss_idx.type))
die(3, "{} != {}: self.data[{t!r}].key(i) does not match self.data[{t!r}][i]!".format(
idx,
ss_idx.idx,
t = ss_idx.type ))
if print_msg:
msg('\b\b\b => {}'.format(SeedID.hlc(sid)))
msg(f'\b\b\b => {SeedID.hlc(sid)}')
seed = self.member_type(self,idx,nonce,length=ss_idx.type)
assert seed.sid == sid,'{} != {}: Seed ID mismatch!'.format(seed.sid,sid)
assert seed.sid == sid, f'{seed.sid} != {sid}: Seed ID mismatch!'
return seed
def get_subseed_by_seed_id(self,sid,last_idx=None,print_msg=False):
@ -130,9 +132,9 @@ class SubSeedList(MMGenObject):
def _collision_debug_msg(self,sid,idx,nonce,nonce_desc='nonce',debug_last_share=False):
slen = 'short' if sid in self.data['short'] else 'long'
m1 = 'add_subseed(idx={},{}):'.format(idx,slen)
m1 = f'add_subseed(idx={idx},{slen}):'
if sid == self.parent_seed.sid:
m2 = 'collision with parent Seed ID {},'.format(sid)
m2 = f'collision with parent Seed ID {sid},'
else:
if debug_last_share:
sl = self.debug_last_share_sid_len
@ -140,8 +142,8 @@ class SubSeedList(MMGenObject):
sid = sid[:sl]
else:
colliding_idx = self.data[slen][sid][0]
m2 = 'collision with ID {} (idx={},{}),'.format(sid,colliding_idx,slen)
msg('{:30} {:46} incrementing {} to {}'.format(m1,m2,nonce_desc,nonce+1))
m2 = f'collision with ID {sid} (idx={colliding_idx},{slen}),'
msg(f'{m1:30} {m2:46} incrementing {nonce_desc} to {nonce+1}')
def _generate(self,last_idx=None,last_sid=None):
@ -184,13 +186,13 @@ class SubSeedList(MMGenObject):
fs1 = '{:>18} {:>18}\n'
fs2 = '{i:>7}L: {:8} {i:>7}S: {:8}\n'
hdr = '{:>16} {} ({} bits)\n\n'.format('Parent Seed:',self.parent_seed.sid.hl(),self.parent_seed.bitlen)
hdr = f' Parent Seed: {self.parent_seed.sid.hl()} ({self.parent_seed.bitlen} bits)\n\n'
hdr += fs1.format('Long Subseeds','Short Subseeds')
hdr += fs1.format('-------------','--------------')
sl = self.data['long'].keys
ss = self.data['short'].keys
body = (fs2.format(sl[n-1],ss[n-1],i=n) for n in r.iterate())
body = (fs2.format( sl[n-1], ss[n-1], i=n ) for n in r.iterate())
return hdr + ''.join(body)
@ -219,7 +221,7 @@ class Seed(SeedBase):
def add_share(ss):
if d.byte_len:
assert ss.byte_len == d.byte_len,'Seed length mismatch! {} != {}'.format(ss.byte_len,d.byte_len)
assert ss.byte_len == d.byte_len, f'Seed length mismatch! {ss.byte_len} != {d.byte_len}'
else:
d.byte_len = ss.byte_len
d.ret ^= int(ss.data.hex(),16)
@ -277,8 +279,7 @@ class SeedShareList(SubSeedList):
ms = SeedShareMaster(self,master_idx,nonce)
if ms.sid == parent_seed.sid:
if g.debug_subseed:
m = 'master_share seed ID collision with parent seed, incrementing nonce to {}'
msg(m.format(nonce+1))
msg(f'master_share seed ID collision with parent seed, incrementing nonce to {nonce+1}')
else:
return ms
raise SubSeedNonceRangeExceeded('nonce range exceeded')
@ -314,11 +315,11 @@ class SeedShareList(SubSeedList):
if g.debug_subseed:
A = parent_seed.data
B = self.join().data
assert A == B,'Data mismatch!\noriginal seed: {!r}\nrejoined seed: {!r}'.format(A,B)
assert A == B, f'Data mismatch!\noriginal seed: {A!r}\nrejoined seed: {B!r}'
def get_share_by_idx(self,idx,base_seed=False):
if idx < 1 or idx > self.count:
raise RangeError('{}: share index out of range'.format(idx))
raise RangeError(f'{idx}: share index out of range')
elif idx == self.count:
return self.last_share
elif self.master_share and idx == 1:
@ -364,7 +365,7 @@ class SeedShareBase(MMGenObject):
@property
def fn_stem(self):
pl = self.parent_list
msdata = '_with_master{}'.format(pl.master_share.idx) if pl.master_share else ''
msdata = f'_with_master{pl.master_share.idx}' if pl.master_share else ''
return '{}-{}-{}of{}{}[{}]'.format(
pl.parent_seed.sid,
pl.id_str,
@ -379,7 +380,7 @@ class SeedShareBase(MMGenObject):
def get_desc(self,ui=False):
pl = self.parent_list
mss = ', with master share #{}'.format(pl.master_share.idx) if pl.master_share else ''
mss = f', with master share #{pl.master_share.idx}' if pl.master_share else ''
if ui:
m = ( yellow("(share {} of {} of ")
+ pl.parent_seed.sid.hl()
@ -398,10 +399,17 @@ class SeedShare(SeedShareBase,SubSeed):
assert parent_list.have_short == False
assert length == 'long'
# field maximums: id_str: none (256 chars), count: 65535 (1024), idx: 65535 (1024), nonce: 65535 (1000)
scramble_key = '{}:{}:'.format(parent_list.split_type,parent_list.id_str).encode() + \
parent_list.count.to_bytes(2,'big') + idx.to_bytes(2,'big') + nonce.to_bytes(2,'big')
scramble_key = (
f'{parent_list.split_type}:{parent_list.id_str}:'.encode() +
parent_list.count.to_bytes(2,'big') +
idx.to_bytes(2,'big') +
nonce.to_bytes(2,'big')
)
if parent_list.master_share:
scramble_key += b':master:' + parent_list.master_share.idx.to_bytes(2,'big')
scramble_key += (
b':master:' +
parent_list.master_share.idx.to_bytes(2,'big')
)
return scramble_seed(seed.data,scramble_key)[:seed.byte_len]
class SeedShareLast(SeedShareBase,SeedBase):
@ -443,7 +451,7 @@ class SeedShareMaster(SeedBase,SeedShareBase):
return '{}-MASTER{}[{}]'.format(
self.parent_list.parent_seed.sid,
self.idx,
self.sid)
self.sid )
def make_base_seed_bin(self):
seed = self.parent_list.parent_seed
@ -460,7 +468,7 @@ class SeedShareMaster(SeedBase,SeedShareBase):
def get_desc(self,ui=False):
psid = self.parent_list.parent_seed.sid
mss = 'master share #{} of '.format(self.idx)
mss = f'master share #{self.idx} of '
return yellow('(' + mss) + psid.hl() + yellow(')') if ui else mss + psid
class SeedShareMasterJoining(SeedShareMaster):

2
mmgen/share/Opts.py
View file

@ -34,7 +34,7 @@ def make_usage_str(prog_name,caller,data):
def gen():
ulbl = 'USAGE:'
for line in lines:
yield '{:{w}} {} {}'.format(ulbl,prog_name,line,w=col1_w)
yield f'{ulbl:{col1_w}} {prog_name} {line}'
ulbl = ''
return ('\n'+(' '*indent)).join(gen())

93
mmgen/tool.py
View file

@ -28,7 +28,7 @@ from .addr import *
NL = ('\n','\r\n')[g.platform=='win']
def _options_annot_str(l):
return "(valid options: '{}')".format("','".join(l))
return "(valid options: '{}')".format( "','".join(l) )
def _create_argtuple(method,localvars):
co = method.__code__
@ -59,10 +59,11 @@ def _create_call_sig(cmd,parsed=False):
c_kwargs = [(a,dfls[n]) for n,a in enumerate(args[nargs:])]
return c_args,dict(c_kwargs),'STDIN_OK' if c_args and ann[args[0]] == 'sstr' else flag
else:
c_args = ['{} [{}]'.format(a,get_type_from_ann(a)) for a in args[:nargs]]
c_args = [f'{a} [{get_type_from_ann(a)}]' for a in args[:nargs]]
c_kwargs = ['"{}" [{}={!r}{}]'.format(
a, type(dfls[n]).__name__, dfls[n],
(' ' + ann[a] if a in ann else ''))
a,
type(dfls[n]).__name__, dfls[n],
(' ' + ann[a] if a in ann else '') )
for n,a in enumerate(args[nargs:])]
return ' '.join(c_args + c_kwargs)
@ -100,7 +101,7 @@ def _usage(cmd=None,exit_val=1):
Msg(' {}{}\n'.format(cls_info[0].upper(),cls_info[1:]))
max_w = max(map(len,bc.user_commands))
for cmd in sorted(bc.user_commands):
Msg(' {:{w}} {}'.format(cmd,_create_call_sig(cmd),w=max_w))
Msg(f' {cmd:{max_w}} {_create_call_sig(cmd)}')
Msg('')
Msg(m2)
elif cmd in MMGenToolCmds:
@ -112,7 +113,7 @@ def _usage(cmd=None,exit_val=1):
_create_call_sig(cmd))
)
else:
die(1,"'{}': no such tool command".format(cmd))
die(1,f'{cmd!r}: no such tool command')
sys.exit(exit_val)
@ -122,8 +123,7 @@ def _process_args(cmd,cmd_args):
if flag != 'VAR_ARGS':
if len(cmd_args) < len(c_args):
m1 = 'Command requires exactly {} non-keyword argument{}'
msg(m1.format(len(c_args),suf(c_args)))
msg(f'Command requires exactly {len(c_args)} non-keyword argument{suf(c_args)}')
_usage(cmd)
u_args = cmd_args[:len(c_args)]
@ -138,9 +138,9 @@ def _process_args(cmd,cmd_args):
u_args[0] = os.read(0,max_dlen)
have_stdin_input = True
if len(u_args[0]) >= max_dlen:
die(2,'Maximum data input for this command is {}'.format(max_dlen_spec))
die(2,f'Maximum data input for this command is {max_dlen_spec}')
if not u_args[0]:
die(2,'{}: ERROR: no output from previous command in pipe'.format(cmd))
die(2,f'{cmd}: ERROR: no output from previous command in pipe')
u_nkwargs = len(cmd_args) - len(c_args)
u_kwargs = {}
@ -149,21 +149,21 @@ def _process_args(cmd,cmd_args):
tk = [a[0] for a in t]
tk_bad = [a for a in tk if a not in c_kwargs]
if set(tk_bad) != set(tk[:len(tk_bad)]): # permit non-kw args to contain '='
die(1,"'{}': illegal keyword argument".format(tk_bad[-1]))
die(1,f'{tk_bad[-1]!r}: illegal keyword argument')
u_kwargs = dict(t[len(tk_bad):])
u_args = cmd_args[:-len(u_kwargs) or None]
elif u_nkwargs > 0:
u_kwargs = dict([a.split('=',1) for a in cmd_args[len(c_args):] if '=' in a])
if len(u_kwargs) != u_nkwargs:
msg('Command requires exactly {} non-keyword argument{}'.format(len(c_args),suf(c_args)))
msg(f'Command requires exactly {len(c_args)} non-keyword argument{suf(c_args)}')
_usage(cmd)
if len(u_kwargs) > len(c_kwargs):
msg('Command accepts no more than {} keyword argument{}'.format(len(c_kwargs),suf(c_kwargs)))
msg(f'Command accepts no more than {len(c_kwargs)} keyword argument{suf(c_kwargs)}')
_usage(cmd)
for k in u_kwargs:
if k not in c_kwargs:
msg("'{}': invalid keyword argument".format(k))
msg(f'{k!r}: invalid keyword argument')
_usage(cmd)
def conv_type(arg,arg_name,arg_type):
@ -179,13 +179,13 @@ def _process_args(cmd,cmd_args):
if arg.lower() in ('true','yes','1','on'): arg = True
elif arg.lower() in ('false','no','0','off'): arg = False
else:
msg("'{}': invalid boolean value for keyword argument".format(arg))
msg(f'{arg!r}: invalid boolean value for keyword argument')
_usage(cmd)
try:
return __builtins__[arg_type](arg)
except:
die(1,"'{}': Invalid argument for argument {} ('{}' required)".format(arg,arg_name,arg_type))
die(1,f'{arg!r}: Invalid argument for argument {arg_name} ({arg_type!r} required)')
if flag == 'VAR_ARGS':
args = [conv_type(u_args[i],c_args[0][0],c_args[0][1]) for i in range(len(u_args))]
@ -223,7 +223,7 @@ def _process_result(ret,pager=False,print_result=False):
# don't add NL to binary data if it can't be converted to utf8
return ret if not print_result else os.write(1,ret)
else:
ydie(1,"tool.py: can't handle return value of type '{}'".format(type(ret).__name__))
ydie(1,f'tool.py: can’t handle return value of type {type(ret).__name__!r}')
from .obj import MMGenAddrType
@ -534,9 +534,9 @@ class MMGenToolCmdCoin(MMGenToolCmds):
def redeem_script2addr(self,redeem_scripthex:'sstr'): # new
"convert a Segwit P2SH-P2WPKH redeem script to an address"
assert self.mmtype.name == 'segwit','This command is meaningful only for --type=segwit'
assert redeem_scripthex[:4] == '0014','{!r}: invalid redeem script'.format(redeem_scripthex)
assert len(redeem_scripthex) == 44,'{} bytes: invalid redeem script length'.format(len(redeem_scripthex)//2)
assert self.mmtype.name == 'segwit', 'This command is meaningful only for --type=segwit'
assert redeem_scripthex[:4] == '0014', f'{redeem_scripthex!r}: invalid redeem script'
assert len(redeem_scripthex) == 44, f'{len(redeem_scripthex)//2} bytes: invalid redeem script length'
return self.pubhash2addr(hash160(redeem_scripthex))
def pubhash2addr(self,pubhashhex:'sstr'):
@ -588,8 +588,9 @@ class MMGenToolCmdMnemonic(MMGenToolCmds):
from .protocol import init_proto
proto = init_proto('xmr')
if len(bytestr) != proto.privkey_len:
m = '{!r}: invalid bit length for Monero private key (must be {})'
die(1,m.format(len(bytestr*8),proto.privkey_len*8))
die(1,'{!r}: invalid bit length for Monero private key (must be {})'.format(
len(bytestr*8),
proto.privkey_len*8 ))
return proto.preprocess_key(bytestr,None)
def _do_random_mn(self,nbytes:int,fmt:str):
@ -598,7 +599,7 @@ class MMGenToolCmdMnemonic(MMGenToolCmds):
if fmt == 'xmrseed':
randbytes = self._xmr_reduce(randbytes)
if opt.verbose:
msg('Seed: {}'.format(randbytes.hex()))
msg(f'Seed: {randbytes.hex()}')
return self.hex2mn(randbytes.hex(),fmt=fmt)
def mn_rand128(self, fmt:mn_opts_disp = dfl_mnemonic_fmt ):
@ -650,7 +651,7 @@ class MMGenToolCmdMnemonic(MMGenToolCmds):
conv_cls = mnemonic_fmts[fmt]['conv_cls']()
ret = conv_cls.get_wordlist(fmt)
if enum:
ret = ['{:>4} {}'.format(n,e) for n,e in enumerate(ret)]
ret = [f'{n:>4} {e}' for n,e in enumerate(ret)]
return '\n'.join(ret)
class MMGenToolCmdFile(MMGenToolCmds):
@ -738,7 +739,7 @@ class MMGenToolCmdFileCrypt(MMGenToolCmds):
data = get_data_from_file(infile,'data for encryption',binary=True)
enc_d = mmgen_encrypt(data,'user data',hash_preset)
if not outfile:
outfile = '{}.{}'.format(os.path.basename(infile),g.mmenc_ext)
outfile = f'{os.path.basename(infile)}.{g.mmenc_ext}'
write_data_to_file(outfile,enc_d,'encrypted data',binary=True)
return True
@ -767,20 +768,22 @@ class MMGenToolCmdFileUtil(MMGenToolCmds):
f = os.open(filename,flgs)
for ch in incog_id:
if ch not in '0123456789ABCDEF':
die(2,"'{}': invalid Incog ID".format(incog_id))
die(2,f'{incog_id!r}: invalid Incog ID')
while True:
d = os.read(f,bsize)
if not d: break
d = carry + d
for i in range(bsize):
if sha256(d[i:i+ivsize]).hexdigest()[:8].upper() == incog_id:
if n+i < ivsize: continue
msg('\rIncog data for ID {} found at offset {}'.format(incog_id,n+i-ivsize))
if not keep_searching: sys.exit(0)
if n+i < ivsize:
continue
msg(f'\rIncog data for ID {incog_id} found at offset {n+i-ivsize}')
if not keep_searching:
sys.exit(0)
carry = d[len(d)-ivsize:]
n += bsize
if not n % mod:
msg_r('\rSearched: {} bytes'.format(n))
msg_r(f'\rSearched: {n} bytes')
msg('')
os.close(f)
@ -826,7 +829,7 @@ class MMGenToolCmdFileUtil(MMGenToolCmds):
blk_size = 1024 * 1024
for i in range(nbytes // blk_size):
if not i % 4:
msg_r('\rRead: {} bytes'.format(i * blk_size))
msg_r(f'\rRead: {i * blk_size} bytes')
q1.put(os.urandom(blk_size))
if nbytes % blk_size:
@ -838,11 +841,11 @@ class MMGenToolCmdFileUtil(MMGenToolCmds):
fsize = os.stat(outfile).st_size
if fsize != nbytes:
die(3,'{}: incorrect random file size (should be {})'.format(fsize,nbytes))
die(3,f'{fsize}: incorrect random file size (should be {nbytes})')
if not silent:
msg('\rRead: {} bytes'.format(nbytes))
qmsg("\r{} byte{} of random data written to file '{}'".format(nbytes,suf(nbytes),outfile))
msg(f'\rRead: {nbytes} bytes')
qmsg(f'\r{nbytes} byte{suf(nbytes)} of random data written to file {outfile!r}')
return True
@ -872,10 +875,10 @@ class MMGenToolCmdWallet(MMGenToolCmds):
return Wallet(sf).seed.subseeds.format(*SubSeedIdxRange(subseed_idx_range))
def list_shares(self,
share_count:int,
id_str='default',
master_share:"(min:1, max:{}, 0=no master share)".format(MasterShareIdx.max_val)=0,
wallet=''):
share_count: int,
id_str = 'default',
master_share: f'(min:1, max:{MasterShareIdx.max_val}, 0=no master share)' = 0,
wallet = '' ):
"list the Seed IDs of the shares resulting from a split of default or specified wallet"
opt.quiet = True
sf = get_seed_file([wallet] if wallet else [],1)
@ -894,8 +897,7 @@ class MMGenToolCmdWallet(MMGenToolCmds):
from .wallet import Wallet
ss = Wallet(sf)
if ss.seed.sid != addr.sid:
m = 'Seed ID of requested address ({}) does not match wallet ({})'
die(1,m.format(addr.sid,ss.seed.sid))
die(1,f'Seed ID of requested address ({addr.sid}) does not match wallet ({ss.seed.sid})')
al = AddrList(
proto = self.proto,
seed = ss.seed,
@ -955,14 +957,15 @@ class MMGenToolCmdRPC(MMGenToolCmds):
sort = set(sort.split(','))
sort_params = {'reverse','age'}
if not sort.issubset(sort_params):
die(1,"The sort option takes the following parameters: '{}'".format("','".join(sort_params)))
die(1,"The sort option takes the following parameters: '{}'".format( "','".join(sort_params) ))
usr_addr_list = []
if mmgen_addrs:
a = mmgen_addrs.rsplit(':',1)
if len(a) != 2:
m = "'{}': invalid address list argument (must be in form <seed ID>:[<type>:]<idx list>)"
die(1,m.format(mmgen_addrs))
die(1,
f'{mmgen_addrs}: invalid address list argument ' +
'(must be in form <seed ID>:[<type>:]<idx list>)' )
usr_addr_list = [MMGenID(self.proto,f'{a[0]}:{i}') for i in AddrIdxList(a[1])]
al = await TwAddrList(self.proto,usr_addr_list,minconf,showempty,showbtcaddrs,all_labels)
@ -1007,7 +1010,7 @@ class MMGenToolCmdRPC(MMGenToolCmds):
from .tw import TrackingWallet
ret = await (await TrackingWallet(self.proto,mode='w')).remove_address(mmgen_or_coin_addr) # returns None on failure
if ret:
msg("Address '{}' deleted from tracking wallet".format(ret))
msg(f'Address {ret!r} deleted from tracking wallet')
return ret
class tool_api(
@ -1101,7 +1104,7 @@ class tool_api(
a description. The first-listed is the default
"""
for t in [MMGenAddrType(proto=self.proto,id_str=id_str) for id_str in self.proto.mmtypes]:
print('{:<12} - {}'.format(t.name,t.desc))
print(f'{t.name:<12} - {t.desc}')
@property
def addrtype(self):

29
mmgen/tw.py
View file

@ -29,7 +29,9 @@ from .tx import is_mmgen_id,is_coin_addr
from .rpc import rpc_init
CUR_HOME,ERASE_ALL = '\033[H','\033[0J'
def CUR_RIGHT(n): return '\033[{}C'.format(n)
def CUR_RIGHT(n):
return f'\033[{n}C'
def get_tw_label(proto,s):
"""
@ -213,7 +215,7 @@ Actions: [q]uit view, [p]rint to file, pager [v]iew, [w]ide view, add [l]abel:
'addr': lambda i: i.addr,
'age': lambda i: 0 - i.confs,
'amt': lambda i: i.amt,
'txid': lambda i: '{} {:04}'.format(i.txid,i.vout),
'txid': lambda i: f'{i.txid} {i.vout:04}',
'twmmid': lambda i: i.twmmid.sort_key
}
key = key or self.sort_key
@ -296,16 +298,21 @@ Actions: [q]uit view, [p]rint to file, pager [v]iew, [w]ide view, add [l]abel:
n = 'Num',
t = 'TXid'.ljust(c.tx_w - 2) + ' Vout',
a = 'Address'.ljust(c.addr_w),
A = 'Amt({})'.format(self.proto.dcoin).ljust(self.disp_prec+5),
A2 = ' Amt({})'.format(self.proto.coin).ljust(self.disp_prec+4),
A = f'Amt({self.proto.dcoin})'.ljust(self.disp_prec+5),
A2 = f' Amt({self.proto.coin})'.ljust(self.disp_prec+4),
c = date_hdr[self.age_fmt],
).rstrip()
for n,i in enumerate(unsp):
addr_dots = '|' + '.'*(c.addr_w-1)
mmid_disp = MMGenID.fmtc('.'*c.mmid_w if i.skip=='addr'
else i.twmmid if i.twmmid.type=='mmgen'
else 'Non-{}'.format(g.proj_name),width=c.mmid_w,color=True)
mmid_disp = MMGenID.fmtc(
(
'.'*c.mmid_w if i.skip == 'addr' else
i.twmmid if i.twmmid.type == 'mmgen' else
f'Non-{g.proj_name}'
),
width = c.mmid_w,
color = True )
if self.show_mmid:
addr_out = '{} {}{}'.format((
@ -354,8 +361,8 @@ Actions: [q]uit view, [p]rint to file, pager [v]iew, [w]ide view, add [l]abel:
t = 'Tx ID,Vout',
a = 'Address'.ljust(addr_w),
m = 'MMGen ID'.ljust(mmid_w),
A = 'Amount({})'.format(self.proto.dcoin),
A2 = 'Amount({})'.format(self.proto.coin),
A = f'Amount({self.proto.dcoin})',
A2 = f'Amount({self.proto.coin})',
c = 'Confs', # skipped for eth
b = 'Block', # skipped for eth
D = 'Date',
@ -713,7 +720,7 @@ class TrackingWallet(MMGenObject,metaclass=AsyncInit):
async def __init__(self,proto,mode='r',token_addr=None):
assert mode in ('r','w','i'), "{!r}: wallet mode must be 'r','w' or 'i'".format(mode)
assert mode in ('r','w','i'), f"{mode!r}: wallet mode must be 'r','w' or 'i'"
if mode == 'i':
self.importing = True
mode = 'w'
@ -892,7 +899,7 @@ class TrackingWallet(MMGenObject,metaclass=AsyncInit):
if self.orig_data != wdata:
if g.debug:
print_stack_trace('TW DATA CHANGED {!r}'.format(self))
print_stack_trace(f'TW DATA CHANGED {self!r}')
print_diff(self.orig_data,wdata,from_json=True)
self.write_changed(wdata)
elif g.debug:

15
mmgen/tx.py
View file

@ -60,7 +60,7 @@ def strfmt_locktime(num,terse=False):
elif num > 0:
return '{}{}'.format(('block height ','')[terse],num)
else:
die(2,"'{}': invalid nLockTime value!".format(num))
die(2,f'{num!r}: invalid nLockTime value!')
def mmaddr2coinaddr(mmaddr,ad_w,ad_f,proto):
@ -83,7 +83,7 @@ def mmaddr2coinaddr(mmaddr,ad_w,ad_f,proto):
def addr2pubhash(proto,addr):
ap = proto.parse_addr(addr)
assert ap,'coin address {!r} could not be parsed'.format(addr)
assert ap,f'coin address {addr!r} could not be parsed'
return ap.bytes.hex()
def addr2scriptPubKey(proto,addr):
@ -101,7 +101,7 @@ def scriptPubKey2addr(proto,s):
elif len(s) == 44 and s[:4] == proto.witness_vernum_hex + '14':
return proto.pubhash2bech32addr(s[4:]),'bech32'
else:
raise NotImplementedError('Unknown scriptPubKey ({})'.format(s))
raise NotImplementedError(f'Unknown scriptPubKey ({s})')
class DeserializedTX(dict,MMGenObject):
"""
@ -159,7 +159,7 @@ class DeserializedTX(dict,MMGenObject):
if has_witness:
u = bshift(2,skip=True).hex()
if u != '0001':
raise IllegalWitnessFlagValue("'{}': Illegal value for flag in transaction!".format(u))
raise IllegalWitnessFlagValue(f'{u!r}: Illegal value for flag in transaction!')
d['num_txins'] = readVInt()
@ -1021,8 +1021,9 @@ class MMGenTX:
def format_view_body(self,blockcount,nonmm_str,max_mmwid,enl,terse,sort):
if sort not in self.view_sort_orders:
die(1,f'{sort!r}: invalid transaction view sort order. Valid options: {{}}'.format(
','.join(self.view_sort_orders) ))
die(1,'{!r}: invalid transaction view sort order. Valid options: {}'.format(
sort,
','.join(self.view_sort_orders) ))
def format_io(desc):
io = getattr(self,desc)
@ -1323,7 +1324,7 @@ class MMGenTX:
uh = dtx['unsigned_hex']
if str(self.txid) != make_chksum_6(bytes.fromhex(uh)).upper():
raise TxHexMismatch('MMGen TxID ({}) does not match hex transaction data!\n{}'.format(self.txid,m))
raise TxHexMismatch(f'MMGen TxID ({self.txid}) does not match hex transaction data!\n{m}')
def compare_size_and_estimated_size(self,tx_decoded):
est_vsize = self.estimate_size()

6
mmgen/txfile.py
View file

@ -45,9 +45,9 @@ class MMGenTxFile:
ymsg('Warning: transaction data appears to be in old format')
import re
d = literal_eval(re.sub(r"[A-Za-z]+?\(('.+?')\)",r'\1',raw_data))
assert type(d) == list,'{} data not a list!'.format(desc)
assert type(d) == list, f'{desc} data not a list!'
if not (desc == 'outputs' and tx.proto.base_coin == 'ETH'): # ETH txs can have no outputs
assert len(d),'no {}!'.format(desc)
assert len(d), f'no {desc}!'
for e in d:
e['amt'] = tx.proto.coin_amt(e['amt'])
io,io_list = (
@ -167,7 +167,7 @@ class MMGenTxFile:
tx.send_amt,
tx.timestamp,
tx.blockcount,
('',' LT={}'.format(tx.locktime))[bool(tx.locktime)]
(f' LT={tx.locktime}' if tx.locktime else ''),
),
tx.hex,
ascii([amt_to_str(e._asdict()) for e in tx.inputs]),

11
mmgen/txsign.py
View file

@ -88,7 +88,7 @@ def add_keys(tx,src,infiles=None,saved_seeds=None,keyaddr_list=None):
for e in need_keys:
for kal in d:
for f in kal.data:
mmid = '{}:{}'.format(kal.al_id,f.idx)
mmid = f'{kal.al_id}:{f.idx}'
if mmid == e.mmid:
if f.addr == e.addr:
e.have_wif = True
@ -151,10 +151,11 @@ async def txsign(tx,seed_files,kl,kal,tx_num_str=''):
tmp.add_wifs(kl)
m = tmp.list_missing('sec')
if m:
die(2, fmt(f"""
ERROR: a key file must be supplied for the following non-{g.proj_name} address{suf(m,'es')}:
{{}}
""".format('\n '.join(m)),strip_char='\t').strip())
die(2, fmt("""
ERROR: a key file must be supplied for the following non-{} address{}:
{{}}
""".format( g.proj_name, suf(m,'es'), '\n '.join(m) ),
strip_char='\t').strip() )
keys += tmp.data
if opt.mmgen_keys_from_file:

6
mmgen/util.py
View file

@ -230,7 +230,7 @@ def parse_bytespec(nbytes):
else:
return int(nbytes)
die(1,"'{}': invalid byte specifier".format(nbytes))
die(1,f'{nbytes!r}: invalid byte specifier')
def check_or_create_dir(path):
try:
@ -290,7 +290,7 @@ def suf(arg,suf_type='s',verb='none'):
elif isinstance(arg,(list,tuple,set,dict)):
n = len(arg)
else:
die(2,'{}: invalid parameter for suf()'.format(arg))
die(2,f'{arg}: invalid parameter for suf()')
return suf_types[verb][suf_type][n == 1]
def get_extension(fn):
@ -795,7 +795,7 @@ def my_raw_input(prompt,echo=True,insert_txt='',use_readline=True):
def keypress_confirm(prompt,default_yes=False,verbose=False,no_nl=False,complete_prompt=False):
q = ('(y/N)','(Y/n)')[bool(default_yes)]
p = prompt if complete_prompt else '{} {}: '.format(prompt,q)
p = prompt if complete_prompt else f'{prompt} {q}: '
nl = ('\n','\r{}\r'.format(' '*len(p)))[no_nl]
if g.accept_defaults:

151
mmgen/wallet.py
View file

@ -30,7 +30,7 @@ from .seed import Seed
def check_usr_seed_len(seed_len):
if opt.seed_len and opt.seed_len != seed_len:
die(1,f"ERROR: requested seed length ({opt.seed_len}) doesn't match seed length of source ({seed_len})")
die(1,f'ERROR: requested seed length ({opt.seed_len}) doesn’t match seed length of source ({seed_len})')
def _is_mnemonic(s,fmt):
oq_save = bool(opt.quiet)
@ -80,7 +80,7 @@ class Wallet(MMGenObject,metaclass=WalletMeta):
if hasattr(opt,'out_fmt') and opt.out_fmt:
out_cls = cls.fmt_code_to_type(opt.out_fmt)
if not out_cls:
die(1,'{!r}: unrecognized output format'.format(opt.out_fmt))
die(1,f'{opt.out_fmt!r}: unrecognized output format')
else:
out_cls = None
@ -148,12 +148,15 @@ class Wallet(MMGenObject,metaclass=WalletMeta):
self._decrypt_retry()
else:
if not self.stdin_ok:
die(1,'Reading from standard input not supported for {} format'.format(self.desc))
die(1,f'Reading from standard input not supported for {self.desc} format')
self._deformat_retry()
self._decrypt_retry()
m = ('',', seed length {}'.format(self.seed.bitlen))[self.seed.bitlen!=256]
qmsg('Valid {} for Seed ID {}{}'.format(self.desc,self.seed.sid.hl(),m))
qmsg('Valid {} for Seed ID {}{}'.format(
self.desc,
self.seed.sid.hl(),
(f', seed length {self.seed.bitlen}' if self.seed.bitlen != 256 else '')
))
def _get_data(self):
if hasattr(self,'infile'):
@ -212,7 +215,7 @@ class Wallet(MMGenObject,metaclass=WalletMeta):
for c in cls.wallet_classes
if hasattr(c,'fmt_codes')]
w = max(len(i[0]) for i in d)
ret = ['{:<{w}} {:<9} {}'.format(a,b,c,w=w) for a,b,c in [
ret = [f'{a:<{w}} {b:<9} {c}' for a,b,c in [
('Format','FileExt','Valid codes'),
('------','-------','-----------')
] + sorted(d)]
@ -260,9 +263,10 @@ class WalletUnenc(Wallet):
msg_r(('\r','\n')[g.test_suite] + ' '*len(prompt) + '\r')
return ok_lens[int(r)-1]
m1 = blue('{} type:'.format(capfirst(desc)))
m2 = yellow(subtype)
msg('{} {}'.format(m1,m2))
msg('{} {}'.format(
blue(f'{capfirst(desc)} type:'),
yellow(subtype)
))
while True:
usr_len = choose_len()
@ -311,7 +315,7 @@ class WalletEnc(Wallet):
else: # Prompt, using old value as default
hp = self._get_hash_preset_from_user(old_hp,add_desc)
if (not opt.keep_hash_preset) and self.op == 'pwchg_new':
qmsg('Hash preset {}'.format('unchanged' if hp==old_hp else f'changed to {hp!r}'))
qmsg('Hash preset {}'.format( 'unchanged' if hp == old_hp else f'changed to {hp!r}' ))
elif opt.hash_preset:
hp = opt.hash_preset
qmsg(f'Using hash preset {hp!r} requested on command line')
@ -379,7 +383,7 @@ class WalletEnc(Wallet):
else:
pw = self._get_new_passphrase()
if self.op == 'pwchg_new':
qmsg('Passphrase {}'.format('unchanged' if pw==old_pw else 'changed'))
qmsg('Passphrase {}'.format( 'unchanged' if pw == old_pw else 'changed' ))
else:
self._get_new_passphrase()
@ -434,13 +438,14 @@ class Mnemonic(WalletUnenc):
mn = self.fmt_data.split()
if len(mn) not in self.mn_lens:
m = 'Invalid mnemonic ({} words). Valid numbers of words: {}'
msg(m.format(len(mn),', '.join(map(str,self.mn_lens))))
msg('Invalid mnemonic ({} words). Valid numbers of words: {}'.format(
len(mn),
', '.join(map(str,self.mn_lens)) ))
return False
for n,w in enumerate(mn,1):
if w not in self.conv_cls.digits[self.wl_id]:
msg('Invalid mnemonic: word #{} is not in the {} wordlist'.format(n,self.wl_id.upper()))
msg(f'Invalid mnemonic: word #{n} is not in the {self.wl_id.upper()} wordlist')
return False
hexseed = self.conv_cls.tohex(mn,self.wl_id,self._mn2hex_pad(mn))
@ -492,27 +497,29 @@ class MMGenSeedFile(WalletUnenc):
b58seed = baseconv.frombytes(self.seed.data,'b58',pad='seed',tostr=True)
self.ssdata.chksum = make_chksum_6(b58seed)
self.ssdata.b58seed = b58seed
self.fmt_data = '{} {}\n'.format(self.ssdata.chksum,split_into_cols(4,b58seed))
self.fmt_data = '{} {}\n'.format(
self.ssdata.chksum,
split_into_cols(4,b58seed) )
def _deformat(self):
desc = self.desc
ld = self.fmt_data.split()
if not (7 <= len(ld) <= 12): # 6 <= padded b58 data (ld[1:]) <= 11
msg('Invalid data length ({}) in {}'.format(len(ld),desc))
msg(f'Invalid data length ({len(ld)}) in {desc}')
return False
a,b = ld[0],''.join(ld[1:])
if not is_chksum_6(a):
msg("'{}': invalid checksum format in {}".format(a, desc))
msg(f'{a!r}: invalid checksum format in {desc}')
return False
if not is_b58_str(b):
msg("'{}': not a base 58 string, in {}".format(b, desc))
msg(f'{b!r}: not a base 58 string, in {desc}')
return False
vmsg_r('Validating {} checksum...'.format(desc))
vmsg_r(f'Validating {desc} checksum...')
if not compare_chksums(a,'file',make_chksum_6(b),'computed',verbose=True):
return False
@ -520,7 +527,7 @@ class MMGenSeedFile(WalletUnenc):
ret = baseconv.tobytes(b,'b58',pad='seed')
if ret == False:
msg('Invalid base-58 encoded seed: {}'.format(val))
msg(f'Invalid base-58 encoded seed: {val}')
return False
self.seed = Seed(ret)
@ -556,8 +563,10 @@ class DieRollSeedFile(WalletUnenc):
rmap = self.conv_cls.seedlen_map_rev['b6d']
if not len(d) in rmap:
m = '{!r}: invalid length for {} (must be one of {})'
raise SeedLengthError(m.format(len(d),self.desc,list(rmap)))
raise SeedLengthError('{!r}: invalid length for {} (must be one of {})'.format(
len(d),
self.desc,
list(rmap) ))
# truncate seed to correct length, discarding high bits
seed_len = rmap[len(d)]
@ -595,7 +604,7 @@ class DieRollSeedFile(WalletUnenc):
b6d_digits = self.conv_cls.digits['b6d']
cr = '\n' if g.test_suite else '\r'
prompt_fs = '\b\b\b {}Enter die roll #{{}}: {}'.format(cr,CUR_SHOW)
prompt_fs = f'\b\b\b {cr}Enter die roll #{{}}: {CUR_SHOW}'
clear_line = '' if g.test_suite else '\r' + ' ' * 25
invalid_msg = CUR_HIDE + cr + 'Invalid entry' + ' ' * 11
@ -638,11 +647,11 @@ class PlainHexSeedFile(WalletUnenc):
d = self.fmt_data.strip()
if not is_hex_str_lc(d):
msg("'{}': not a lowercase hexadecimal string, in {}".format(d,desc))
msg(f'{d!r}: not a lowercase hexadecimal string, in {desc}')
return False
if not len(d)*4 in g.seed_lens:
msg('Invalid data length ({}) in {}'.format(len(d),desc))
msg(f'Invalid data length ({len(d)}) in {desc}')
return False
self.seed = Seed(bytes.fromhex(d))
@ -663,7 +672,9 @@ class MMGenHexSeedFile(WalletUnenc):
h = self.seed.hexdata
self.ssdata.chksum = make_chksum_6(h)
self.ssdata.hexseed = h
self.fmt_data = '{} {}\n'.format(self.ssdata.chksum, split_into_cols(4,h))
self.fmt_data = '{} {}\n'.format(
self.ssdata.chksum,
split_into_cols(4,h) )
def _deformat(self):
desc = self.desc
@ -672,22 +683,22 @@ class MMGenHexSeedFile(WalletUnenc):
d[1]
chk,hstr = d[0],''.join(d[1:])
except:
msg("'{}': invalid {}".format(self.fmt_data.strip(),desc))
msg(f'{self.fmt_data.strip()!r}: invalid {desc}')
return False
if not len(hstr)*4 in g.seed_lens:
msg('Invalid data length ({}) in {}'.format(len(hstr),desc))
msg(f'Invalid data length ({len(hstr)}) in {desc}')
return False
if not is_chksum_6(chk):
msg("'{}': invalid checksum format in {}".format(chk, desc))
msg(f'{chk!r}: invalid checksum format in {desc}')
return False
if not is_hex_str(hstr):
msg("'{}': not a hexadecimal string, in {}".format(hstr, desc))
msg(f'{hstr!r}: not a hexadecimal string, in {desc}')
return False
vmsg_r('Validating {} checksum...'.format(desc))
vmsg_r(f'Validating {desc} checksum...')
if not compare_chksums(chk,'file',make_chksum_6(hstr),'computed',verbose=True):
return False
@ -738,17 +749,17 @@ class MMGenWallet(WalletEnc):
old_lbl = self.ss_in.ssdata.label
if opt.keep_label:
lbl = old_lbl
qmsg('Reusing label {} at user request'.format(lbl.hl(encl="''")))
qmsg('Reusing label {} at user request'.format( lbl.hl(encl="''") ))
elif self.label:
lbl = self.label
qmsg('Using label {} requested on command line'.format(lbl.hl(encl="''")))
qmsg('Using label {} requested on command line'.format( lbl.hl(encl="''") ))
else: # Prompt, using old value as default
lbl = self._get_label_from_user(old_lbl)
if (not opt.keep_label) and self.op == 'pwchg_new':
qmsg('Label {}'.format('unchanged' if lbl==old_lbl else f'changed to {lbl!r}'))
qmsg('Label {}'.format( 'unchanged' if lbl == old_lbl else f'changed to {lbl!r}' ))
elif self.label:
lbl = self.label
qmsg('Using label {} requested on command line'.format(lbl.hl(encl="''")))
qmsg('Using label {} requested on command line'.format( lbl.hl(encl="''") ))
else:
lbl = self._get_label_from_user()
self.ssdata.label = lbl
@ -767,11 +778,10 @@ class MMGenWallet(WalletEnc):
es_fmt = baseconv.frombytes(d.enc_seed,'b58',pad='seed',tostr=True)
lines = (
d.label,
'{} {} {} {} {}'.format(s.sid.lower(), d.key_id.lower(),
s.bitlen, d.pw_status, d.timestamp),
'{}: {} {} {}'.format(d.hash_preset,*get_hash_params(d.hash_preset)),
'{} {}'.format(make_chksum_6(slt_fmt),split_into_cols(4,slt_fmt)),
'{} {}'.format(make_chksum_6(es_fmt), split_into_cols(4,es_fmt))
'{} {} {} {} {}'.format( s.sid.lower(), d.key_id.lower(), s.bitlen, d.pw_status, d.timestamp ),
'{}: {} {} {}'.format( d.hash_preset, *get_hash_params(d.hash_preset) ),
'{} {}'.format( make_chksum_6(slt_fmt), split_into_cols(4,slt_fmt) ),
'{} {}'.format( make_chksum_6(es_fmt), split_into_cols(4,es_fmt) )
)
chksum = make_chksum_6(' '.join(lines).encode())
self.fmt_data = '\n'.join((chksum,)+lines) + '\n'
@ -781,11 +791,11 @@ class MMGenWallet(WalletEnc):
def check_master_chksum(lines,desc):
if len(lines) != 6:
msg('Invalid number of lines ({}) in {} data'.format(len(lines),desc))
msg(f'Invalid number of lines ({len(lines)}) in {desc} data')
return False
if not is_chksum_6(lines[0]):
msg('Incorrect master checksum ({}) in {} data'.format(lines[0],desc))
msg(f'Incorrect master checksum ({lines[0]}) in {desc} data')
return False
chk = make_chksum_6(' '.join(lines[1:]))
@ -811,14 +821,14 @@ class MMGenWallet(WalletEnc):
hpdata = lines[3].split()
d.hash_preset = hp = hpdata[0][:-1] # a string!
qmsg("Hash preset of wallet: '{}'".format(hp))
qmsg(f'Hash preset of wallet: {hp!r}')
if opt.hash_preset and opt.hash_preset != hp:
qmsg('Warning: ignoring user-requested hash preset {opt.hash_preset}')
hash_params = tuple(map(int,hpdata[1:]))
if hash_params != get_hash_params(d.hash_preset):
msg(f"Hash parameters {' '.join(hash_params)!r} don't match hash preset {d.hash_preset!r}")
msg(f'Hash parameters {" ".join(hash_params)!r} don’t match hash preset {d.hash_preset!r}')
return False
lmin,foo,lmax = sorted(baseconv.seedlen_map_rev['b58']) # 22,33,44
@ -828,7 +838,7 @@ class MMGenWallet(WalletEnc):
b58_val = ''.join(l)
if len(b58_val) < lmin or len(b58_val) > lmax:
msg('Invalid format for {} in {}: {}'.format(key,self.desc,l))
msg(f'Invalid format for {key} in {self.desc}: {l}')
return False
if not compare_chksums(chk,key,
@ -837,7 +847,7 @@ class MMGenWallet(WalletEnc):
val = baseconv.tobytes(b58_val,'b58',pad='seed')
if val == False:
msg('Invalid base 58 number: {}'.format(b58_val))
msg(f'Invalid base 58 number: {b58_val}')
return False
setattr(d,key,val)
@ -907,7 +917,7 @@ class Brainwallet(WalletEnc):
buflen = bw_seed_len // 8 )
qmsg('Done')
self.seed = Seed(seed)
msg('Seed ID: {}'.format(self.seed.sid))
msg(f'Seed ID: {self.seed.sid}')
qmsg('Check this value against your records')
return True
@ -971,7 +981,7 @@ to exit and re-run the program with the '--old-incog-fmt' option.
# IV is used BOTH to initialize counter and to salt password!
d.iv = get_random(g.aesctr_iv_len)
d.iv_id = self._make_iv_chksum(d.iv)
msg('New Incog Wallet ID: {}'.format(d.iv_id))
msg(f'New Incog Wallet ID: {d.iv_id}')
qmsg('Make a record of this value')
vmsg(self.msg['record_incog_id'])
@ -982,7 +992,7 @@ to exit and re-run the program with the '--old-incog-fmt' option.
d.wrapper_key = make_key(d.passwd, d.iv, d.hash_preset, 'incog wrapper key')
d.key_id = make_chksum_8(d.wrapper_key)
vmsg('Key ID: {}'.format(d.key_id))
vmsg(f'Key ID: {d.key_id}')
d.target_data_len = self._get_incog_data_len(self.seed.bitlen)
def _format(self):
@ -1010,7 +1020,7 @@ to exit and re-run the program with the '--old-incog-fmt' option.
d.iv = self.fmt_data[0:g.aesctr_iv_len]
d.incog_id = self._make_iv_chksum(d.iv)
d.enc_incog_data = self.fmt_data[g.aesctr_iv_len:]
msg('Incog Wallet ID: {}'.format(d.incog_id))
msg(f'Incog Wallet ID: {d.incog_id}')
qmsg('Check this value against your records')
vmsg(self.msg['check_incog_id'])
@ -1019,14 +1029,14 @@ to exit and re-run the program with the '--old-incog-fmt' option.
def _verify_seed_newfmt(self,data):
chk,seed = data[:8],data[8:]
if sha256(seed).digest()[:8] == chk:
qmsg('Passphrase{} are correct'.format(self.msg['dec_chk'].format('and')))
qmsg('Passphrase{} are correct'.format( self.msg['dec_chk'].format('and') ))
return seed
else:
msg('Incorrect passphrase{}'.format(self.msg['dec_chk'].format('or')))
msg('Incorrect passphrase{}'.format( self.msg['dec_chk'].format('or') ))
return False
def _verify_seed_oldfmt(self,seed):
m = 'Seed ID: {}. Is the Seed ID correct?'.format(make_chksum_8(seed))
m = f'Seed ID: {make_chksum_8(seed)}. Is the Seed ID correct?'
if keypress_confirm(m, True):
return seed
else:
@ -1045,7 +1055,7 @@ to exit and re-run the program with the '--old-incog-fmt' option.
d.enc_seed = dd[g.salt_len:]
key = make_key(d.passwd, d.salt, d.hash_preset, 'main key')
qmsg('Key ID: {}'.format(make_chksum_8(key)))
qmsg(f'Key ID: {make_chksum_8(key)}')
verify_seed = getattr(self,'_verify_seed_'+
('newfmt','oldfmt')[bool(opt.old_incog_fmt)])
@ -1054,7 +1064,7 @@ to exit and re-run the program with the '--old-incog-fmt' option.
if seed:
self.seed = Seed(seed)
msg('Seed ID: {}'.format(self.seed.sid))
msg(f'Seed ID: {self.seed.sid}')
return True
else:
return False
@ -1123,14 +1133,19 @@ harder to find, you're advised to choose a much larger file size than this.
d = self.ssdata
m = ('Input','Destination')[action=='write']
if fn.size < d.hincog_offset + d.target_data_len:
fs = "{} file '{}' has length {}, too short to {} {} bytes of data at offset {}"
die(1,fs.format(m,fn.name,fn.size,action,d.target_data_len,d.hincog_offset))
die(1,'{} file {!r} has length {}, too short to {} {} bytes of data at offset {}'.format(
m,
fn.name,
fn.size,
action,
d.target_data_len,
d.hincog_offset ))
def _get_data(self):
d = self.ssdata
d.hincog_offset = self._get_hincog_params('input')[1]
qmsg("Getting hidden incog data from file '{}'".format(self.infile.name))
qmsg(f'Getting hidden incog data from file {self.infile.name!r}')
# Already sanity-checked:
d.target_data_len = self._get_incog_data_len(opt.seed_len or g.dfl_seed_len)
@ -1141,7 +1156,7 @@ harder to find, you're advised to choose a much larger file size than this.
os.lseek(fh,int(d.hincog_offset),os.SEEK_SET)
self.fmt_data = os.read(fh,d.target_data_len)
os.close(fh)
qmsg("Data read from file '{}' at offset {}".format(self.infile.name,d.hincog_offset))
qmsg(f'Data read from file {self.infile.name!r} at offset {d.hincog_offset}')
# overrides method in Wallet
def write_to_file(self):
@ -1160,14 +1175,16 @@ harder to find, you're advised to choose a much larger file size than this.
try:
os.stat(fn)
except:
if keypress_confirm("Requested file '{}' does not exist. Create?".format(fn),default_yes=True):
if keypress_confirm(
f'Requested file {fn!r} does not exist. Create?',
default_yes = True ):
min_fsize = d.target_data_len + d.hincog_offset
msg(self.msg['choose_file_size'].format(min_fsize))
while True:
fsize = parse_bytespec(my_raw_input('Enter file size: '))
if fsize >= min_fsize:
break
msg('File size must be an integer no less than {}'.format(min_fsize))
msg(f'File size must be an integer no less than {min_fsize}')
from .tool import MMGenToolCmdFileUtil
MMGenToolCmdFileUtil().rand2file(fn,str(fsize))
@ -1178,16 +1195,22 @@ harder to find, you're advised to choose a much larger file size than this.
from .filename import Filename
f = Filename(fn,ftype=type(self),write=True)
dmsg('{} data len {}, offset {}'.format(capfirst(self.desc),d.target_data_len,d.hincog_offset))
dmsg('{} data len {}, offset {}'.format(
capfirst(self.desc),
d.target_data_len,
d.hincog_offset ))
if check_offset:
self._check_valid_offset(f,'write')
if not opt.quiet:
confirm_or_raise('',"alter file '{}'".format(f.name))
confirm_or_raise( '', f'alter file {f.name!r}' )
flgs = os.O_RDWR|os.O_BINARY if g.platform == 'win' else os.O_RDWR
fh = os.open(f.name,flgs)
os.lseek(fh, int(d.hincog_offset), os.SEEK_SET)
os.write(fh, self.fmt_data)
os.close(fh)
msg("{} written to file '{}' at offset {}".format(capfirst(self.desc),f.name,d.hincog_offset))
msg('{} written to file {!r} at offset {}'.format(
capfirst(self.desc),
f.name,
d.hincog_offset ))

19
mmgen/xmrwallet.py
View file

@ -32,7 +32,7 @@ from .obj import CoinAddr,CoinTxID,SeedID,AddrIdx,Hilite,InitErrors
xmrwallet_uarg_info = (
lambda e,hp: {
'daemon': e('HOST:PORT', hp),
'tx_relay_daemon': e('HOST:PORT[:PROXY_HOST:PROXY_PORT]', r'({p})(?::({p}))?'.format(p=hp)),
'tx_relay_daemon': e('HOST:PORT[:PROXY_HOST:PROXY_PORT]', rf'({hp})(?::({hp}))?'),
'transfer_spec': e('SOURCE_WALLET_NUM:ACCOUNT:ADDRESS,AMOUNT', rf'(\d+):(\d+):([{_b58a}]+),([0-9.]+)'),
'sweep_spec': e('SOURCE_WALLET_NUM:ACCOUNT[,DEST_WALLET_NUM]', r'(\d+):(\d+)(?:,(\d+))?'),
})(
@ -53,7 +53,7 @@ class XMRWalletAddrSpec(str,Hilite,InitErrors,MMGenObject):
try:
if isinstance(arg1,str):
me = str.__new__(cls,arg1)
m = re.fullmatch('({n}):({n}):({n}|None)'.format(n=r'[0-9]{1,4}'),arg1)
m = re.fullmatch( '({n}):({n}):({n}|None)'.format(n=r'[0-9]{1,4}'), arg1 )
assert m is not None, f'{arg1!r}: invalid XMRWalletAddrSpec'
for e in m.groups():
if len(e) != 1 and e[0] == '0':
@ -371,7 +371,10 @@ class MoneroWalletOps:
'-α' if g.debug_utf8 else '' ))
async def main(self):
gmsg('\n{}ing {} wallet{}'.format(self.desc,len(self.addr_data),suf(self.addr_data)))
gmsg('\n{}ing {} wallet{}'.format(
self.desc,
len(self.addr_data),
suf(self.addr_data) ))
processed = 0
for n,d in enumerate(self.addr_data): # [d.sec,d.addr,d.wallet_passwd,d.viewkey]
fn = self.get_wallet_fn(d)
@ -382,7 +385,7 @@ class MoneroWalletOps:
os.path.basename(fn),
))
processed += await self.process_wallet(d,fn)
gmsg('\n{} wallet{} {}'.format(processed,suf(processed),self.past))
gmsg(f'\n{processed} wallet{suf(processed)} {self.past}')
return processed
class rpc:
@ -578,7 +581,7 @@ class MoneroWalletOps:
restore_height = uopt.restore_height,
language = 'English' )
pp_msg(ret) if opt.debug else msg(' Address: {}'.format(ret['address']))
pp_msg(ret) if opt.debug else msg(' Address: {}'.format( ret['address'] ))
return True
class sync(wallet):
@ -651,8 +654,10 @@ class MoneroWalletOps:
self.accts_data[bn] = { 'accts': a, 'addrs': b }
msg(' Wallet height: {}'.format(wallet_height))
msg(' Sync time: {:02}:{:02}'.format( t_elapsed//60, t_elapsed%60 ))
msg(f' Wallet height: {wallet_height}')
msg(' Sync time: {:02}:{:02}'.format(
t_elapsed // 60,
t_elapsed % 60 ))
await self.c.call('close_wallet')
return wallet_height >= chain_height

8
setup.py
View file

@ -43,10 +43,10 @@ class my_build_ext(build_ext):
setup(
cmdclass = { 'build_ext': my_build_ext },
ext_modules = [Extension(
name = 'mmgen.secp256k1',
sources = ['extmod/secp256k1mod.c'],
libraries = ([],['gmp'])[have_msys2],
name = 'mmgen.secp256k1',
sources = ['extmod/secp256k1mod.c'],
libraries = ([],['gmp'])[have_msys2],
extra_objects = [os.path.join(ext_path,'.libs/libsecp256k1.a')],
include_dirs = [os.path.join(ext_path,'include')],
include_dirs = [os.path.join(ext_path,'include')],
)]
)