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

util.py: relocate lesser-used functions to util2.py

Browse source
This commit is contained in:
The MMGen Project 2022-10-17 18:37:22 +00:00
commit 9888fe4c65
Signed by: mmgen
GPG key ID: 3F8B1861E32B7DA2
21 changed files with 192 additions and 173 deletions
Download patch file Download diff file Expand all files Collapse all files

2
mmgen/addrgen.py
View file

@ -46,7 +46,7 @@ class addr_generator:
def __init__(self,proto,addr_type):
super().__init__(proto,addr_type)
from .util import get_keccak
from .util2 import get_keccak
self.keccak_256 = get_keccak()
def AddrGenerator(proto,addr_type):

7
mmgen/crypto.py
View file

@ -38,7 +38,7 @@ from .util import (
get_words_from_user,
make_chksum_8,
compare_chksums,
pwfile_reuse_warning,
oneshot_warning,
)
mmenc_ext = 'mmenc'
@ -63,6 +63,11 @@ hash_presets = {
'7': _hp(18, 8, 24),
}
class pwfile_reuse_warning(oneshot_warning):
message = 'Reusing passphrase from file {!r} at user request'
def __init__(self,fn):
oneshot_warning.__init__(self,div=fn,fmt_args=[fn],reverse=True)
def get_hash_params(hash_preset):
if hash_preset in hash_presets:
return hash_presets[hash_preset] # N,r,p

3
mmgen/main_tool.py
View file

@ -229,6 +229,7 @@ def process_args(cmd,cmd_args,cls):
if sys.stdin.isatty():
die( 'BadFilename', "Standard input is a TTY. Can't use '-' as a filename" )
else:
from .util2 import parse_bytespec
max_dlen_spec = '10kB' # limit input to 10KB for now
max_dlen = parse_bytespec(max_dlen_spec)
u_args[0] = os.read(0,max_dlen)
@ -293,7 +294,7 @@ def process_args(cmd,cmd_args,cls):
return ( args, kwargs )
def process_result(ret,pager=False,print_result=False):
from .util import Msg,die,parse_bytespec
from .util import Msg,die
"""
Convert result to something suitable for output to screen and return it.
If result is bytes and not convertible to utf8, output as binary using os.write().

4
mmgen/proto/eth/contract.py
View file

@ -155,7 +155,7 @@ class Token(TokenCommon):
def __init__(self,proto,addr,decimals,rpc=None):
if type(self).__name__ == 'Token':
from ...util import get_keccak
from ...util2 import get_keccak
self.keccak_256 = get_keccak()
self.proto = proto
self.addr = TokenAddr(proto,addr)
@ -167,7 +167,7 @@ class Token(TokenCommon):
class TokenResolve(TokenCommon,metaclass=AsyncInit):
async def __init__(self,proto,rpc,addr):
from ...util import get_keccak
from ...util2 import get_keccak
self.keccak_256 = get_keccak()
self.proto = proto
self.rpc = rpc

3
mmgen/proto/eth/misc.py
View file

@ -12,7 +12,8 @@
proto.eth.misc: miscellaneous utilities for Ethereum base protocol
"""
from ...util import die,get_keccak
from ...util import die
from ...util2 import get_keccak
def extract_key_from_geth_keystore_wallet(wallet_fn,passwd,check_addr=True):
"""

2
mmgen/proto/eth/pyethereum/utils.py
View file

@ -6,7 +6,7 @@ from py_ecc.secp256k1 import privtopub,ecdsa_raw_sign,ecdsa_raw_recover
from .. import rlp
from ..rlp.sedes import Binary
from ....util import get_keccak
from ....util2 import get_keccak
keccak_256 = get_keccak()
def sha3_256(bstr):

2
mmgen/proto/xmr/keygen.py
View file

@ -24,7 +24,7 @@ class backend:
from ...proto.xmr.params import mainnet
self.proto_cls = mainnet
from ...util import get_keccak
from ...util2 import get_keccak
self.keccak_256 = get_keccak()
def to_viewkey(self,privkey):

2
mmgen/protocol.py
View file

@ -95,7 +95,7 @@ class CoinProtocol(MMGenObject):
assert self.name.startswith('Ethereum'), 'CoinProtocol.Base_chk1'
if self.base_coin in ('ETH','XMR'):
from .util import get_keccak
from .util2 import get_keccak
self.keccak_256 = get_keccak()
if need_amt:

3
mmgen/tool/fileutil.py
View file

@ -23,7 +23,7 @@ tool/fileutil.py: File routines for the 'mmgen-tool' utility
import os
from .common import tool_cmd_base
from ..util import msg,msg_r,qmsg,die,suf,parse_bytespec,make_full_path
from ..util import msg,msg_r,qmsg,die,suf,make_full_path
from ..crypto import get_random,aesctr_iv_len
class tool_cmd(tool_cmd_base):
@ -95,6 +95,7 @@ class tool_cmd(tool_cmd_base):
from cryptography.hazmat.backends import default_backend
from ..opts import opt
from ..util2 import parse_bytespec
def encrypt_worker(wid):
ctr_init_val = os.urandom( aesctr_iv_len )

3
mmgen/tool/help.py
View file

@ -25,7 +25,8 @@ import mmgen.main_tool as main_tool
def main_help():
from ..util import pretty_format,capfirst
from ..util import capfirst
from ..util2 import pretty_format
def do():
for clsname,cmdlist in main_tool.mods.items():

16
mmgen/tool/util.py
View file

@ -25,7 +25,7 @@ from .common import tool_cmd_base
class tool_cmd(tool_cmd_base):
"general string conversion and hashing utilities"
# mmgen.util.bytespec_map
# mmgen.util2.bytespec_map
def bytespec(self,dd_style_byte_specifier:str):
"""
convert a byte specifier such as ‘4GB’ into an integer
@ -48,10 +48,10 @@ class tool_cmd(tool_cmd_base):
EB = 1000000000000000000
E = 1152921504606846976
"""
from ..util import parse_bytespec
from ..util2 import parse_bytespec
return parse_bytespec(dd_style_byte_specifier)
# mmgen.util.bytespec_map
# mmgen.util2.bytespec_map
def to_bytespec(self,
n: int,
dd_style_byte_specifier: str,
@ -78,7 +78,7 @@ class tool_cmd(tool_cmd_base):
EB = 1000000000000000000
E = 1152921504606846976
"""
from ..util import int2bytespec
from ..util2 import int2bytespec
return int2bytespec( n, dd_style_byte_specifier, fmt, print_sym )
def randhex(self,
@ -107,7 +107,7 @@ class tool_cmd(tool_cmd_base):
line_nums: "format for line numbers (valid choices: 'hex','dec')" = 'hex'):
"create hexdump of data from file (use '-' for stdin)"
from ..fileutil import get_data_from_file
from ..util import pretty_hexdump
from ..util2 import pretty_hexdump
data = get_data_from_file( infile, dash=True, quiet=True, binary=True )
return pretty_hexdump( data, cols=cols, line_nums=line_nums ).rstrip()
@ -118,7 +118,7 @@ class tool_cmd(tool_cmd_base):
import sys,os,msvcrt
msvcrt.setmode( sys.stdout.fileno(), os.O_BINARY )
from ..fileutil import get_data_from_file
from ..util import decode_pretty_hexdump
from ..util2 import decode_pretty_hexdump
hexdata = get_data_from_file( infile, dash=True, quiet=True )
return decode_pretty_hexdump(hexdata)
@ -138,7 +138,7 @@ class tool_cmd(tool_cmd_base):
from ..fileutil import get_data_from_file
b = get_data_from_file( data, binary=True )
elif hex_input:
from ..util import decode_pretty_hexdump
from ..util2 import decode_pretty_hexdump
b = decode_pretty_hexdump(data)
else:
b = data
@ -219,7 +219,7 @@ class tool_cmd(tool_cmd_base):
add_spaces: 'add a space after every 5th character' = True):
"convert a hexadecimal string to die roll base6 (base6d)"
from ..baseconv import baseconv
from ..util import block_format
from ..util2 import block_format
ret = baseconv('b6d').fromhex(hexstr,pad,tostr=True)
return block_format( ret, gw=5, cols=None ).strip() if add_spaces else ret

151
mmgen/util.py
View file

@ -17,7 +17,7 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
"""
util.py: Low-level routines imported by other modules in the MMGen suite
util.py: Frequently-used variables, classes and functions for the MMGen suite
"""
import sys,os,time,re
@ -26,8 +26,6 @@ from .color import *
from .globalvars import g
from .opts import opt
import mmgen.color as color_mod
ascii_lowercase = 'abcdefghijklmnopqrstuvwxyz'
hexdigits = '0123456789abcdefABCDEF'
@ -136,21 +134,6 @@ def die(ev,s='',stdout=False):
else:
raise ValueError(f'{ev}: exit value must be string or int instance')
def die_wait(delay,ev=0,s=''):
assert isinstance(delay,int)
assert isinstance(ev,int)
if s:
msg(s)
time.sleep(delay)
sys.exit(ev)
def die_pause(ev=0,s=''):
assert isinstance(ev,int)
if s:
msg(s)
input('Press ENTER to exit')
sys.exit(ev)
def Die(ev=0,s=''):
die(ev=ev,s=s,stdout=True)
@ -191,12 +174,6 @@ def list_gen(*data):
yield i[0]
return list(gen())
def removeprefix(s,pfx): # workaround for pre-Python 3.9
return s[len(pfx):] if s.startswith(pfx) else s
def removesuffix(s,sfx): # workaround for pre-Python 3.9
return s[:len(sfx)] if s.endswith(sfx) else s
def remove_dups(iterable,edesc='element',desc='list',quiet=False,hide=False):
"""
Remove duplicate occurrences of iterable elements, preserving first occurrence
@ -215,70 +192,6 @@ def exit_if_mswin(feature):
if g.platform == 'win':
die(2, capfirst(feature) + ' not supported on the MSWin / MSYS2 platform' )
def get_keccak(cached_ret=[]):
if not cached_ret:
from .opts import opt
# called in opts.init() via CoinProtocol, so must use getattr():
if getattr(opt,'use_internal_keccak_module',False):
qmsg('Using internal keccak module by user request')
from .contrib.keccak import keccak_256
else:
try:
from sha3 import keccak_256
except:
from .contrib.keccak import keccak_256
cached_ret.append(keccak_256)
return cached_ret[0]
# From 'man dd':
# c=1, w=2, b=512, kB=1000, K=1024, MB=1000*1000, M=1024*1024,
# GB=1000*1000*1000, G=1024*1024*1024, and so on for T, P, E, Z, Y.
bytespec_map = (
('c', 1),
('w', 2),
('b', 512),
('kB', 1000),
('K', 1024),
('MB', 1000000),
('M', 1048576),
('GB', 1000000000),
('G', 1073741824),
('TB', 1000000000000),
('T', 1099511627776),
('PB', 1000000000000000),
('P', 1125899906842624),
('EB', 1000000000000000000),
('E', 1152921504606846976),
)
def int2bytespec(n,spec,fmt,print_sym=True):
def spec2int(spec):
for k,v in bytespec_map:
if k == spec:
return v
else:
die('{spec}: unrecognized bytespec')
return '{:{}f}{}'.format( n / spec2int(spec), fmt, spec if print_sym else '' )
def parse_bytespec(nbytes):
m = re.match(r'([0123456789.]+)(.*)',nbytes)
if m:
if m.group(2):
for k,v in bytespec_map:
if k == m.group(2):
from decimal import Decimal
return int(Decimal(m.group(1)) * v)
else:
msg("Valid byte specifiers: '{}'".format("' '".join([i[0] for i in bytespec_map])))
elif '.' in nbytes:
raise ValueError('fractional bytes not allowed')
else:
return int(nbytes)
die(1,f'{nbytes!r}: invalid byte specifier')
def suf(arg,suf_type='s',verb='none'):
suf_types = {
'none': {
@ -338,10 +251,6 @@ def make_chksum_6(s):
def is_chksum_6(s):
return len(s) == 6 and set(s) <= set(hexdigits_lc)
def make_iv_chksum(s):
from hashlib import sha256
return sha256(s).hexdigest()[:8].upper()
def split_into_cols(col_wid,s):
return ' '.join([s[col_wid*i:col_wid*(i+1)] for i in range(len(s)//col_wid+1)]).rstrip()
@ -377,19 +286,6 @@ def secs_to_hms(secs):
def secs_to_ms(secs):
return '{:02d}:{:02d}'.format(secs//60, secs % 60)
def format_elapsed_hr(t,now=None,cached={}):
e = int((now or time.time()) - t)
if not e in cached:
abs_e = abs(e)
cached[e] = ', '.join(
'{} {}{}'.format(n,desc,suf(n)) for desc,n in (
('day', abs_e // 86400),
('hour', abs_e // 3600 % 24),
('minute', abs_e // 60 % 60),
) if n
) + (' ago' if e > 0 else ' in the future') if abs_e // 60 else 'just now'
return cached[e]
def is_int(s):
try:
int(str(s))
@ -411,43 +307,6 @@ def is_utf8(s):
def remove_whitespace(s,ws='\t\r\n '):
return s.translate(dict((ord(e),None) for e in ws))
def pretty_format(s,width=80,pfx=''):
out = []
while(s):
if len(s) <= width:
out.append(s)
break
i = s[:width].rfind(' ')
out.append(s[:i])
s = s[i+1:]
return pfx + ('\n'+pfx).join(out)
def block_format(data,gw=2,cols=8,line_nums=None,data_is_hex=False):
assert line_nums in (None,'hex','dec'),"'line_nums' must be one of None, 'hex' or 'dec'"
ln_fs = '{:06x}: ' if line_nums == 'hex' else '{:06}: '
bytes_per_chunk = gw
if data_is_hex:
gw *= 2
nchunks = len(data)//gw + bool(len(data)%gw)
return ''.join(
('' if (line_nums == None or i % cols) else ln_fs.format(i*bytes_per_chunk))
+ data[i*gw:i*gw+gw]
+ (' ' if (not cols or (i+1) % cols) else '\n')
for i in range(nchunks)
).rstrip() + '\n'
def pretty_hexdump(data,gw=2,cols=8,line_nums=None):
return block_format(data.hex(),gw,cols,line_nums,data_is_hex=True)
def decode_pretty_hexdump(data):
pat = re.compile(fr'^[{hexdigits}]+:\s+')
lines = [pat.sub('',line) for line in data.splitlines()]
try:
return bytes.fromhex(''.join((''.join(lines).split())))
except:
msg('Data not in hexdump format')
return False
def strip_comment(line):
return re.sub('#.*','',line).rstrip()
@ -511,8 +370,9 @@ class oneshot_warning:
def do(self,wcls,div,fmt_args,reverse):
def do_warning():
import mmgen.color
message = getattr(wcls,'message')
color = getattr( color_mod, getattr(wcls,'color') )
color = getattr( mmgen.color, getattr(wcls,'color') )
msg(color('WARNING: ' + message.format(*fmt_args)))
if not hasattr(wcls,'data'):
@ -535,11 +395,6 @@ class oneshot_warning_group(oneshot_warning):
def __init__(self,wcls,div=None,fmt_args=[],reverse=False):
self.do(getattr(self,wcls),div,fmt_args,reverse)
class pwfile_reuse_warning(oneshot_warning):
message = 'Reusing passphrase from file {!r} at user request'
def __init__(self,fn):
oneshot_warning.__init__(self,div=fn,fmt_args=[fn],reverse=True)
def line_input(prompt,echo=True,insert_txt=''):
"""
multi-line prompts OK

151
mmgen/util2.py Executable file
View file

@ -0,0 +1,151 @@
#!/usr/bin/env python3
#
# mmgen = Multi-Mode GENerator, a command-line cryptocurrency wallet
# Copyright (C)2013-2022 The MMGen Project <mmgen@tuta.io>
# Licensed under the GNU General Public License, Version 3:
# https://www.gnu.org/licenses
# Public project repositories:
# https://github.com/mmgen/mmgen
# https://gitlab.com/mmgen/mmgen
"""
util2.py: Less frequently-used variables, classes and utility functions for the MMGen suite
"""
import re,time
from .util import msg,qmsg,suf,hexdigits
def die_wait(delay,ev=0,s=''):
assert isinstance(delay,int)
assert isinstance(ev,int)
if s:
msg(s)
time.sleep(delay)
sys.exit(ev)
def die_pause(ev=0,s=''):
assert isinstance(ev,int)
if s:
msg(s)
input('Press ENTER to exit')
sys.exit(ev)
def removeprefix(s,pfx): # workaround for pre-Python 3.9
return s[len(pfx):] if s.startswith(pfx) else s
def removesuffix(s,sfx): # workaround for pre-Python 3.9
return s[:len(sfx)] if s.endswith(sfx) else s
def get_keccak(cached_ret=[]):
if not cached_ret:
from .opts import opt
# called in opts.init() via CoinProtocol, so must use getattr():
if getattr(opt,'use_internal_keccak_module',False):
qmsg('Using internal keccak module by user request')
from .contrib.keccak import keccak_256
else:
try:
from sha3 import keccak_256
except:
from .contrib.keccak import keccak_256
cached_ret.append(keccak_256)
return cached_ret[0]
# From 'man dd':
# c=1, w=2, b=512, kB=1000, K=1024, MB=1000*1000, M=1024*1024,
# GB=1000*1000*1000, G=1024*1024*1024, and so on for T, P, E, Z, Y.
bytespec_map = (
('c', 1),
('w', 2),
('b', 512),
('kB', 1000),
('K', 1024),
('MB', 1000000),
('M', 1048576),
('GB', 1000000000),
('G', 1073741824),
('TB', 1000000000000),
('T', 1099511627776),
('PB', 1000000000000000),
('P', 1125899906842624),
('EB', 1000000000000000000),
('E', 1152921504606846976),
)
def int2bytespec(n,spec,fmt,print_sym=True):
def spec2int(spec):
for k,v in bytespec_map:
if k == spec:
return v
else:
die('{spec}: unrecognized bytespec')
return '{:{}f}{}'.format( n / spec2int(spec), fmt, spec if print_sym else '' )
def parse_bytespec(nbytes):
m = re.match(r'([0123456789.]+)(.*)',nbytes)
if m:
if m.group(2):
for k,v in bytespec_map:
if k == m.group(2):
from decimal import Decimal
return int(Decimal(m.group(1)) * v)
else:
msg("Valid byte specifiers: '{}'".format("' '".join([i[0] for i in bytespec_map])))
elif '.' in nbytes:
raise ValueError('fractional bytes not allowed')
else:
return int(nbytes)
die(1,f'{nbytes!r}: invalid byte specifier')
def format_elapsed_hr(t,now=None,cached={}):
e = int((now or time.time()) - t)
if not e in cached:
abs_e = abs(e)
cached[e] = ' '.join(
'{} {}{}'.format(n,desc,suf(n)) for desc,n in (
('day', abs_e // 86400),
('hour', abs_e // 3600 % 24),
('minute', abs_e // 60 % 60),
) if n
) + (' ago' if e > 0 else ' in the future') if abs_e // 60 else 'just now'
return cached[e]
def pretty_format(s,width=80,pfx=''):
out = []
while(s):
if len(s) <= width:
out.append(s)
break
i = s[:width].rfind(' ')
out.append(s[:i])
s = s[i+1:]
return pfx + ('\n'+pfx).join(out)
def block_format(data,gw=2,cols=8,line_nums=None,data_is_hex=False):
assert line_nums in (None,'hex','dec'),"'line_nums' must be one of None, 'hex' or 'dec'"
ln_fs = '{:06x}: ' if line_nums == 'hex' else '{:06}: '
bytes_per_chunk = gw
if data_is_hex:
gw *= 2
nchunks = len(data)//gw + bool(len(data)%gw)
return ''.join(
('' if (line_nums == None or i % cols) else ln_fs.format(i*bytes_per_chunk))
+ data[i*gw:i*gw+gw]
+ (' ' if (not cols or (i+1) % cols) else '\n')
for i in range(nchunks)
).rstrip() + '\n'
def pretty_hexdump(data,gw=2,cols=8,line_nums=None):
return block_format(data.hex(),gw,cols,line_nums,data_is_hex=True)
def decode_pretty_hexdump(data):
pat = re.compile(fr'^[{hexdigits}]+:\s+')
lines = [pat.sub('',line) for line in data.splitlines()]
try:
return bytes.fromhex(''.join((''.join(lines).split())))
except:
msg('Data not in hexdump format')
return False

3
mmgen/wallet/dieroll.py
View file

@ -15,7 +15,8 @@ wallet.dieroll: dieroll wallet class
import time
from ..globalvars import g
from ..opts import opt
from ..util import msg,msg_r,die,fmt,block_format,remove_whitespace,keypress_confirm
from ..util import msg,msg_r,die,fmt,remove_whitespace,keypress_confirm
from ..util2 import block_format
from ..seed import Seed
from ..baseconv import baseconv
from .unenc import wallet

2
mmgen/wallet/incog_hex.py
View file

@ -12,7 +12,7 @@
wallet.incog_hex: hexadecimal incognito wallet class
"""
from ..util import pretty_hexdump,decode_pretty_hexdump
from ..util2 import pretty_hexdump,decode_pretty_hexdump
from .incog_base import wallet
class wallet(wallet):

2
mmgen/wallet/incog_hidden.py
View file

@ -24,11 +24,11 @@ from ..util import (
die,
compare_or_die,
keypress_confirm,
parse_bytespec,
line_input,
capfirst,
confirm_or_raise
)
from ..util2 import parse_bytespec
from .incog_base import wallet
class wallet(wallet):

1
mmgen/xmrwallet.py
View file

@ -142,6 +142,7 @@ class MoneroMMGenTX:
if pmid:
fs += ' Payment ID: {pmid}'
from .util2 import format_elapsed_hr
return fmt(fs,strip_char='\t',indent=indent).format(
orange(self.base_chksum.upper()),
d.seed_id.hl(),

2
scripts/exec_wrapper.py
View file

@ -44,7 +44,7 @@ def exec_wrapper_write_traceback(e,exit_val):
cwd = os.path.abspath('.')
def fixup_fn(fn_in):
from mmgen.util import removeprefix,removesuffix
from mmgen.util2 import removeprefix,removesuffix
fn = removeprefix(removeprefix(fn_in,cwd+'/'),'test/overlay/tree/')
return removesuffix(fn,'_orig.py') + '.py' if fn.endswith('_orig.py') else fn
# Python 3.9:

2
test/misc/oneshot_warning.py
View file

@ -30,6 +30,8 @@ class wg(oneshot_warning_group):
for i in (1,2,3):
from mmgen.crypto import pwfile_reuse_warning
msg('\npw')
for k in ('A','B'):
ret = pwfile_reuse_warning(k).warning_shown

2
test/test.py
View file

@ -376,7 +376,7 @@ class CmdGroupMgr(object):
def gen():
for name,data in cls.cmd_group_in:
if name.startswith('subgroup.'):
from mmgen.util import removeprefix
from mmgen.util2 import removeprefix
sg_key = removeprefix(name,'subgroup.')
# sg_key = name.removeprefix('subgroup.') # Python 3.9
if sg_name in (None,sg_key):

2
test/test_py_d/ts_tool.py
View file

@ -33,7 +33,7 @@ class TestSuiteTool(TestSuiteMain,TestSuiteBase):
def tool_rand2file(self):
outfile = os.path.join(self.tmpdir,'rand2file.out')
from mmgen.util import parse_bytespec
from mmgen.util2 import parse_bytespec
for nbytes in ('1','1023','1K','1048575','1M','1048577','123M'):
t = self.spawn(
'mmgen-tool',