Support UTF-8 filenames and paths

This commit is contained in:
The MMGen Project 2018-03-09 13:16:10 +00:00
commit 896c7fe996
Signed by: mmgen
GPG key ID: 3F8B1861E32B7DA2
19 changed files with 146 additions and 113 deletions

View file

@ -1,4 +1,5 @@
#!/usr/bin/env python
# -*- coding: UTF-8 -*-
#
# mmgen = Multi-Mode GENerator, command-line Bitcoin cold storage solution
# Copyright (C)2013-2018 The MMGen Project <mmgen@tuta.io>
@ -473,7 +474,7 @@ Removed {{}} duplicate WIF key{{}} from keylist (also in {pnm} key-address file
self.ext += '.'+g.mmenc_ext
def write_to_file(self,ask_tty=True,ask_write_default_yes=False,binary=False,desc=None):
fn = u'{}.{}'.format(self.id_str,self.ext)
fn = u'{}{x}.{}'.format(self.id_str,self.ext,x=u'' if g.debug_utf8 else '')
ask_tty = self.has_keys and not opt.quiet
write_data_to_file(fn,self.fmt_data,desc or self.file_desc,ask_tty=ask_tty,binary=binary)

View file

@ -1,4 +1,5 @@
#!/usr/bin/env python
# -*- coding: UTF-8 -*-
#
# mmgen = Multi-Mode GENerator, command-line Bitcoin cold storage solution
# Copyright (C)2013-2018 The MMGen Project <mmgen@tuta.io>
@ -77,7 +78,7 @@ interactive prompt, may be specified as either absolute {} amounts, using
a plain decimal number, or as satoshis per byte, using an integer followed by
the letter 's'.
""".format(g.coin),
'txsign': """
'txsign': u"""
Transactions may contain both {pnm} or non-{pnm} input addresses.
To sign non-{pnm} inputs, a {dn} wallet dump or flat key list is used
@ -107,4 +108,4 @@ column below:
pnm=g.proj_name,
pnu=g.proto.name.capitalize(),
pnl=g.proj_name.lower())
}[k]
}[k] + u'' if g.debug_utf8 else ''

View file

@ -90,8 +90,9 @@ class g(object):
alice = False
# test suite:
bogus_wallet_data = ''
bogus_wallet_data = u''
traceback_cmd = 'scripts/traceback.py'
debug_utf8 = False
for k in ('win','linux'):
if sys.platform[:len(k)] == k:
@ -143,6 +144,7 @@ class g(object):
'MMGEN_DEBUG_OPTS',
'MMGEN_DEBUG_RPC',
'MMGEN_DEBUG_ADDRLIST',
'MMGEN_DEBUG_UTF8',
'MMGEN_QUIET',
'MMGEN_DISABLE_COLOR',
'MMGEN_FORCE_256_COLOR',

View file

@ -84,7 +84,7 @@ opts_data = lambda: {
what=gen_what,g=g,
dmat="'{}' or '{}'".format(g.proto.dfl_mmtype,MAT.mmtypes[g.proto.dfl_mmtype]['name'])
),
'notes': """
'notes': u"""
NOTES FOR THIS COMMAND

View file

@ -118,7 +118,7 @@ msg_fmt = '{{:{}}} {{:34}} {{:{}}}'.format(w_n_of_m,w_mmid)
if opt.rescan: import threading
msg('Importing {} address{} from {}{}'.format(
msg(u'Importing {} address{} from {}{}'.format(
len(al.data),
suf(al.data,'es'),
infile,

View file

@ -70,7 +70,7 @@ opts_data = lambda: {
g=g,pnm=g.proj_name,d58=dfl_len['b58'],d32=dfl_len['b32'],dhex=dfl_len['hex'],
kgs=' '.join(['{}:{}'.format(n,k) for n,k in enumerate(g.key_generators,1)])
),
'notes': """
'notes': u"""
NOTES FOR THIS COMMAND

View file

@ -49,7 +49,7 @@ if not opt.status: do_license_msg()
from mmgen.tx import *
tx = MMGenTX(infile,silent_open=True) # sig check performed here
vmsg("Signed transaction file '{}' is valid".format(infile))
vmsg(u"Signed transaction file '{}' is valid".format(infile))
if not tx.marked_signed():
die(1,'Transaction is not signed!')

View file

@ -98,7 +98,7 @@ for tx_num,tx_file in enumerate(tx_files,1):
if tx.marked_signed():
die(1,'Transaction is already signed!')
vmsg("Successfully opened transaction file '{}'".format(tx_file))
vmsg(u"Successfully opened transaction file '{}'".format(tx_file))
if opt.tx_id:
msg(tx.txid); continue

View file

@ -95,16 +95,16 @@ opts_data = lambda: {
iaction=capfirst(iaction),
oaction=capfirst(oaction),
),
'notes': """
'notes': u"""
{n_pw}{n_bw}
FMT CODES:
{f}
""".format(
f='\n '.join(SeedSource.format_fmt_codes().splitlines()),
f=u'\n '.join(SeedSource.format_fmt_codes().splitlines()),
n_pw=help_notes('passwd'),
n_bw=('','\n\n' + help_notes('brainwallet'))[bw_note]
n_bw=(u'','\n\n' + help_notes('brainwallet'))[bw_note]
)
}

View file

@ -1,4 +1,5 @@
#!/usr/bin/env python
# -*- coding: UTF-8 -*-
#
# mmgen = Multi-Mode GENerator, command-line Bitcoin cold storage solution
# Copyright (C)2013-2018 The MMGen Project <mmgen@tuta.io>
@ -63,12 +64,12 @@ def opt_postproc_debug():
Msg(' Opts after processing:')
for k in a:
v = getattr(opt,k)
Msg(' {:18}: {:<6} [{}]'.format(k,v,type(v).__name__))
Msg(u' {:18}: {:<6} [{}]'.format(k,v,type(v).__name__))
Msg(" Opts set to 'None':")
Msg(' {}\n'.format('\n '.join(b)))
Msg(u' {}\n'.format('\n '.join(b)))
Msg(' Global vars:')
for e in [d for d in dir(g) if d[:2] != '__']:
Msg(' {:<20}: {}'.format(e, getattr(g,e)))
Msg(u' {:<20}: {}'.format(e, getattr(g,e)))
Msg('\n=== end opts.py debug ===\n')
def opt_postproc_initializations():
@ -168,7 +169,7 @@ def override_from_env():
val = os.getenv(name) # os.getenv() returns None if env var is unset
if val: # exclude empty string values too
gname = name[idx:].lower()
setattr(g,gname,set_for_type(val,getattr(g,gname),name,invert_bool))
setattr(g,gname,set_for_type(val.decode('utf8'),getattr(g,gname),name,invert_bool))
def warn_altcoins(trust_level):
if trust_level == None: return
@ -291,6 +292,10 @@ def init(opts_f,add_opts=[],opt_filter=None):
if do_help: # print help screen only after global vars are initialized
opts_data = opts_f()
opts_data['long_options'] = common_opts_data
if g.debug_utf8:
for k in opts_data:
if type(opts_data[k]) in (str,unicode):
opts_data[k] += u''
mmgen.share.Opts.parse_opts(sys.argv,opts_data,opt_filter=opt_filter)
if g.bob or g.alice:

View file

@ -38,7 +38,7 @@ common_args = lambda: (
'--rpcpassword={}'.format(rpc_password),
'--rpcport={}'.format(rpc_port),
'--regtest',
'--datadir={}'.format(data_dir))
u'--datadir={}'.format(data_dir))
def start_daemon(user,quiet=False,daemon=True,reindex=False):
# requires Bitcoin ABC version >= 0.16.2
@ -53,7 +53,7 @@ def start_daemon(user,quiet=False,daemon=True,reindex=False):
) + add_args + common_args()
if daemon: cmd += ('--daemon',)
if reindex: cmd += ('--reindex',)
if not g.debug or quiet: vmsg('{}'.format(' '.join(cmd)))
if not g.debug or quiet: vmsg(u'{}'.format(' '.join(cmd)))
p = subprocess.Popen(cmd,stdout=PIPE,stderr=PIPE)
err = process_output(p,silent=False)[1]
if err:
@ -71,7 +71,7 @@ def start_cmd(*args,**kwargs):
if args[0] == 'cli':
cmd = (g.proto.name+'-cli',) + common_args() + args[1:]
if g.debug or not 'quiet' in kwargs:
vmsg('{}'.format(' '.join(cmd)))
vmsg(u'{}'.format(' '.join(cmd)))
ip = op = ep = (PIPE,None)['no_pipe' in kwargs and kwargs['no_pipe']]
if 'pipe_stdout_only' in kwargs and kwargs['pipe_stdout_only']: ip = ep = None
return subprocess.Popen(cmd,stdin=ip,stdout=op,stderr=ep)

View file

@ -1,4 +1,5 @@
#!/usr/bin/env python
# -*- coding: UTF-8 -*-
#
# mmgen = Multi-Mode GENerator, command-line Bitcoin cold storage solution
# Copyright (C)2013-2018 The MMGen Project <mmgen@tuta.io>
@ -198,11 +199,11 @@ class SeedSource(MMGenObject):
for c in cls.get_subclasses()
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 = [u'{:<{w}} {:<9} {}'.format(a,b,c,w=w) for a,b,c in [
('Format','FileExt','Valid codes'),
('------','-------','-----------')
] + sorted(d)]
return '\n'.join(ret) + '\n'
return u'\n'.join(ret) + ('',u'')[g.debug_utf8] + '\n'
def get_fmt_data(self):
self._format()
@ -227,7 +228,7 @@ class SeedSourceUnenc(SeedSource):
def _encrypt(self): pass
def _filename(self):
return '{}[{}].{}'.format(self.seed.sid,self.seed.length,self.ext)
return u'{}[{}]{x}.{}'.format(self.seed.sid,self.seed.length,self.ext,x=u'' if g.debug_utf8 else '')
class SeedSourceEnc(SeedSource):
@ -314,7 +315,7 @@ an empty passphrase, just hit ENTER twice.
return pw
def _get_passphrase(self,desc_suf=''):
desc ='{}passphrase for {}{}'.format(
desc = u'{}passphrase for {}{}'.format(
('','old ')[self.op=='pwchg_old'],
self.desc,
('',' '+desc_suf)[bool(desc_suf)]
@ -323,7 +324,7 @@ an empty passphrase, just hit ENTER twice.
w = pwfile_reuse_warning()
ret = ' '.join(get_words_from_file(opt.passwd_file,desc,silent=w))
else:
ret = ' '.join(get_words_from_user('Enter {}: '.format(desc)))
ret = ' '.join(get_words_from_user(u'Enter {}: '.format(desc)))
self.ssdata.passwd = ret
def _get_first_pw_and_hp_and_encrypt_seed(self):
@ -713,13 +714,13 @@ class Wallet (SeedSourceEnc):
return False
def _filename(self):
return '{}-{}[{},{}].{}'.format(
return u'{}-{}[{},{}]{x}.{}'.format(
self.seed.sid,
self.ssdata.key_id,
self.seed.length,
self.ssdata.hash_preset,
self.ext
)
self.ext,
x=u'' if g.debug_utf8 else '')
class Brainwallet (SeedSourceEnc):
@ -847,13 +848,14 @@ to exit and re-run the program with the '--old-incog-fmt' option.
def _filename(self):
s = self.seed
d = self.ssdata
return '{}-{}-{}[{},{}].{}'.format(
return u'{}-{}-{}[{},{}]{x}.{}'.format(
s.sid,
d.key_id,
d.iv_id,
s.length,
d.hash_preset,
self.ext)
self.ext,
x=u'' if g.debug_utf8 else '')
def _deformat(self):
@ -974,7 +976,7 @@ harder to find, you're advised to choose a much larger file size than this.
d = self.ssdata
d.hincog_offset = self._get_hincog_params('input')[1]
qmsg("Getting hidden incog data from file '{}'".format(self.infile.name))
qmsg(u"Getting hidden incog data from file '{}'".format(self.infile.name))
# Already sanity-checked:
d.target_data_len = self._get_incog_data_len(opt.seed_len)
@ -985,7 +987,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(u"Data read from file '{}' at offset {}".format(self.infile.name,d.hincog_offset))
# overrides method in SeedSource
def write_to_file(self):
@ -1004,7 +1006,7 @@ 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(u"Requested file '{}' does not exist. Create?".format(fn),default_yes=True):
min_fsize = d.target_data_len + d.hincog_offset
msg(self.msg['choose_file_size'].format(min_fsize))
while True:
@ -1032,4 +1034,4 @@ 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)
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(u"{} written to file '{}' at offset {}".format(capfirst(self.desc),f.name,d.hincog_offset))

View file

@ -30,16 +30,16 @@ def usage(opts_data):
def print_help_and_exit(opts_data,longhelp=False):
pn = opts_data['prog_name']
pn_len = str(len(pn)+2)
out = ' {:<{p}} {}\n'.format(pn.upper()+':',opts_data['desc'].strip(),p=pn_len)
out += ' {:<{p}} {} {}\n'.format('USAGE:',pn,opts_data['usage'].strip(),p=pn_len)
out = u' {:<{p}} {}\n'.format(pn.upper()+':',opts_data['desc'].strip(),p=pn_len)
out += u' {:<{p}} {} {}\n'.format('USAGE:',pn,opts_data['usage'].strip(),p=pn_len)
od_opts = opts_data[('options','long_options')[longhelp]].strip().splitlines()
hdr = ('OPTIONS:',' LONG OPTIONS:')[longhelp]
ls = (' ','')[longhelp]
es = ('',' ')[longhelp]
out += '{ls}{}\n{ls}{es}{}'.format(hdr,('\n'+ls).join(od_opts),ls=ls,es=es)
out += u'{ls}{}\n{ls}{es}{}'.format(hdr,('\n'+ls).join(od_opts),ls=ls,es=es)
if 'notes' in opts_data and not longhelp:
out += '\n ' + '\n '.join(opts_data['notes'][1:-1].splitlines())
print(out)
out += '\n ' + '\n '.join(opts_data['notes'].strip().splitlines())
print(out.encode('utf8'))
sys.exit(0)
def process_opts(argv,opts_data,short_opts,long_opts,defer_help=False):
@ -108,7 +108,7 @@ def parse_opts(argv,opts_data,opt_filter=None,defer_help=False):
if not skip: od[-1][3] += '\n' + l
opts_data[k] = '\n'.join(
['{:<3} --{} {}'.format(
[u'{:<3} --{} {}'.format(
('-'+d[0]+',','')[d[0]=='-'],d[1],d[3]) for d in od if d[6] == False]
)
od_all += od

View file

@ -25,17 +25,24 @@ from binascii import hexlify
from mmgen.common import *
def path_join(*args,**kwargs):
if not 'decode' in kwargs: kwargs['decode'] = True
assert type(kwargs['decode']) == bool
assert kwargs.keys() == ['decode']
ret = os.path.join(*[a.encode('utf8') for a in args])
return ret.decode('utf8') if kwargs['decode'] else ret
def cleandir(d):
from shutil import rmtree
try: files = os.listdir(d)
try: files = [f.decode('utf8') for f in os.listdir(d)]
except: return
gmsg("Cleaning directory '{}'".format(d))
gmsg(u"Cleaning directory '{}'".format(d))
for f in files:
try:
os.unlink(os.path.join(d,f))
os.unlink(path_join(d,f,decode=False))
except:
rmtree(os.path.join(d,f))
rmtree(path_join(d,f,decode=False))
def getrandnum(n): return int(hexlify(os.urandom(n)),16)
def getrandhex(n): return hexlify(os.urandom(n))
@ -49,7 +56,7 @@ def mk_tmpdir(d):
except OSError as e:
if e.errno != 17: raise
else:
vmsg("Created directory '{}'".format(d))
vmsg(u"Created directory '{}'".format(d))
def mk_tmpdir_path(path,cfg):
try:

View file

@ -473,7 +473,7 @@ def Rand2file(outfile,nbytes,threads=4,silent=False):
if not silent:
msg('\rRead: {} bytes'.format(nbytes))
qmsg("\r{} bytes of random data written to file '{}'".format(nbytes,outfile))
qmsg(u"\r{} bytes of random data written to file '{}'".format(nbytes,outfile))
q1.join()
q2.join()
f.close()

View file

@ -1,4 +1,5 @@
#!/usr/bin/env python
# -*- coding: UTF-8 -*-
#
# mmgen = Multi-Mode GENerator, command-line Bitcoin cold storage solution
# Copyright (C)2013-2018 The MMGen Project <mmgen@tuta.io>
@ -855,13 +856,14 @@ class MMGenTX(MMGenObject):
def create_fn(self):
tl = self.get_hex_locktime()
self.fn = '{}{}[{!s}{}{}].{}'.format(
self.fn = u'{}{}[{!s}{}{}]{x}.{}'.format(
self.txid,
('-'+g.coin,'')[g.coin=='BTC'],
self.send_amt,
('',',{}'.format(self.btc2spb(self.get_fee())))[self.is_rbf()],
('',',tl={}'.format(tl))[bool(tl)],
self.ext)
self.ext,
x=u'' if g.debug_utf8 else '')
def write_to_file( self,
add_desc='',

View file

@ -80,7 +80,7 @@ def pdie(*args):
def set_for_type(val,refval,desc,invert_bool=False,src=None):
src_str = (''," in '{}'".format(src))[bool(src)]
if type(refval) == bool:
v = str(val).lower()
v = unicode(val).lower()
if v in ('true','yes','1'): ret = True
elif v in ('false','no','none','0'): ret = False
else: die(1,"'{}': invalid value for '{}'{} (must be of type '{}')".format(
@ -90,7 +90,7 @@ def set_for_type(val,refval,desc,invert_bool=False,src=None):
try:
ret = type(refval)((val,not val)[invert_bool])
except:
die(1,"'{}': invalid value for '{}'{} (must be of type '{}')".format(
die(1,u"'{}': invalid value for '{}'{} (must be of type '{}')".format(
val,desc,src_str,type(refval).__name__))
return ret
@ -116,13 +116,14 @@ def parse_nbytes(nbytes):
die(1,"'{}': invalid byte specifier".format(nbytes))
def check_or_create_dir(path):
path_enc = path.encode('utf8')
try:
os.listdir(path)
os.listdir(path_enc)
except:
try:
os.makedirs(path,0700)
os.makedirs(path_enc,0700)
except:
die(2,"ERROR: unable to read or create path '{}'".format(path))
die(2,u"ERROR: unable to read or create path '{}'".format(path))
from mmgen.opts import opt
@ -451,7 +452,7 @@ def check_file_type_and_access(fname,ftype,blkdev_ok=False):
try: mode = os.stat(fname).st_mode
except:
die(1,"Unable to stat requested {} '{}'".format(ftype,fname))
die(1,u"Unable to stat requested {} '{}'".format(ftype,fname))
for t in ok_types:
if t[0](mode): break
@ -566,7 +567,7 @@ def write_data_to_file(
confirm_or_exit('','output {} to pipe'.format(desc))
msg('')
of2,pd = os.path.relpath(of),os.path.pardir
msg("Redirecting output to file '{}'".format((of2,of)[of2[:len(pd)] == pd]))
msg(u"Redirecting output to file '{}'".format((of2,of)[of2[:len(pd)] == pd]))
else:
msg('Redirecting output to file')
@ -588,16 +589,16 @@ def write_data_to_file(
hush = False
if file_exists(outfile) and ask_overwrite:
q = "File '{}' already exists\nOverwrite?".format(outfile)
q = u"File '{}' already exists\nOverwrite?".format(outfile)
confirm_or_exit('',q)
msg("Overwriting file '{}'".format(outfile))
msg(u"Overwriting file '{}'".format(outfile))
hush = True
f = open_file_or_exit(outfile,('w','wb')[bool(binary)])
try:
f.write(data)
except:
die(2,"Failed to write {} to file '{}'".format(desc,outfile))
die(2,u"Failed to write {} to file '{}'".format(desc,outfile))
f.close
if not (hush or silent):
@ -639,7 +640,7 @@ def mmgen_decrypt_file_maybe(fn,desc='',silent=False):
have_enc_ext = get_extension(fn) == g.mmenc_ext
if have_enc_ext or not is_utf8(d):
m = ('Attempting to decrypt','Decrypting')[have_enc_ext]
msg("{} {} '{}'".format(m,desc,fn))
msg(u"{} {} '{}'".format(m,desc,fn))
from mmgen.crypto import mmgen_decrypt_retry
d = mmgen_decrypt_retry(d,desc)
return d
@ -709,7 +710,7 @@ def keypress_confirm(prompt,default_yes=False,verbose=False,no_nl=False):
from mmgen.term import get_char
q = ('(y/N)','(Y/n)')[bool(default_yes)]
p = '{} {}: '.format(prompt,q)
p = u'{} {}: '.format(prompt,q)
nl = ('\n','\r{}\r'.format(' '*len(p)))[no_nl]
if opt.accept_defaults:

View file

@ -46,7 +46,7 @@ def my_send(p,t,delay=send_delay,s=False):
if opt.verbose:
ls = (' ','')[bool(opt.debug or not s)]
es = (' ','')[bool(s)]
msg('{}SEND {}{}'.format(ls,es,yellow("'{}'"%t.replace('\n',r'\n'))))
msg(u'{}SEND {}{}'.format(ls,es,yellow(u"'{}'".format(t.decode('utf8').replace('\n',r'\n')))))
return ret
def my_expect(p,s,t='',delay=send_delay,regex=False,nonl=False,silent=False):
@ -85,6 +85,8 @@ def debug_pexpect_msg(p):
errmsg('\n{}{}{}'.format(red('BEFORE ['),p.before,red(']')))
errmsg('{}{}{}'.format(red('MATCH ['),p.after,red(']')))
data_dir = os.path.join('test','data_dir'+('',u'')[bool(os.getenv('MMGEN_DEBUG_UTF8'))])
class MMGenPexpect(object):
NL = '\r\n'
@ -97,7 +99,7 @@ class MMGenPexpect(object):
cmd_args = ['--{}{}'.format(k.replace('_','-'),
'='+getattr(opt,k) if getattr(opt,k) != True else ''
) for k in passthru_args if getattr(opt,k)] \
+ ['--data-dir='+os.path.join('test','data_dir')] + cmd_args
+ ['--data-dir='+data_dir] + cmd_args
if g.platform == 'win': cmd,args = 'python',[mmgen_cmd]+cmd_args
else: cmd,args = mmgen_cmd,cmd_args
@ -109,9 +111,9 @@ class MMGenPexpect(object):
die(2,(m1+m2).format(name,cmd,args))
if opt.popen_spawn:
args = [(a,"'{}'".format(a))[' ' in a] for a in args]
args = [u'{q}{}{q}'.format(a,q="'" if ' ' in a else '') for a in args]
cmd_str = '{} {}'.format(cmd,' '.join(args)).replace('\\','/')
cmd_str = u'{} {}'.format(cmd,u' '.join(args)).replace('\\','/')
if opt.coverage:
fs = 'python -m trace --count --coverdir={} --file={} {c}'
cmd_str = fs.format(*init_coverage(),c=cmd_str)
@ -124,7 +126,7 @@ class MMGenPexpect(object):
clr1,clr2,eol = ((green,cyan,'\n'),(nocolor,nocolor,' '))[bool(opt.print_cmdline)]
sys.stderr.write(green('Testing: {}\n'.format(desc)))
if not msg_only:
sys.stderr.write(clr1('Executing {}{}'.format(clr2(cmd_str),eol)))
sys.stderr.write(clr1(u'Executing {}{}'.format(clr2(cmd_str),eol)))
else:
m = 'Testing {}: '.format(desc)
msg_r(m)
@ -144,7 +146,7 @@ class MMGenPexpect(object):
cmd_str = g.traceback_cmd + ' ' + cmd_str
# Msg('\ncmd_str: {}'.format(cmd_str))
if opt.popen_spawn:
self.p = PopenSpawn(cmd_str)
self.p = PopenSpawn(cmd_str.encode('utf8'))
else:
self.p = pexpect.spawn(cmd,args)
if opt.exact_output: self.p.logfile = sys.stdout
@ -220,14 +222,14 @@ class MMGenPexpect(object):
if ret == 1:
my_send(self.p,'YES\n')
# if oo:
outfile = self.expect_getend("Overwriting file '").rstrip("'")
outfile = self.expect_getend("Overwriting file '").rstrip("'").decode('utf8')
return outfile
# else:
# ret = my_expect(self.p,s1)
self.expect(self.NL,nonl=True)
outfile = self.p.before.strip().strip("'")
outfile = self.p.before.strip().strip("'").decode('utf8')
if opt.debug_pexpect: rmsg('Outfile [{}]'.format(outfile))
vmsg('{} file: {}'.format(desc,cyan(outfile.replace("'",''))))
vmsg(u'{} file: {}'.format(desc,cyan(outfile.replace("'",''))))
return outfile
def no_overwrite(self):
@ -248,7 +250,8 @@ class MMGenPexpect(object):
self.expect(self.NL,nonl=True,silent=True)
debug_pexpect_msg(self.p)
end = self.p.before
vmsg(' ==> {}'.format(cyan(end)))
if not g.debug:
vmsg(' ==> {}'.format(cyan(end)))
return end
def interactive(self):

View file

@ -75,21 +75,23 @@ sample_text = \
# under '/dev/shm' and put datadir and temp files here.
shortopts = ''.join([e[1:] for e in sys.argv if len(e) > 1 and e[0] == '-' and e[1] != '-'])
shortopts = ['-'+e for e in list(shortopts)]
data_dir = os.path.join('test','data_dir')
data_dir_basename = 'data_dir' + ('',u'')[bool(os.getenv('MMGEN_DEBUG_UTF8'))]
data_dir = path_join('test',data_dir_basename)
data_dir_enc = data_dir.encode('utf8')
if not any(e in ('--skip-deps','--resume','-S','-r') for e in sys.argv+shortopts):
if g.platform == 'win':
try: os.listdir(data_dir)
try: os.listdir(data_dir_enc)
except: pass
else:
try: shutil.rmtree(data_dir)
try: shutil.rmtree(data_dir_enc)
except: # we couldn't remove data dir - perhaps regtest daemon is running
try: subprocess.call(['python',os.path.join('cmds','mmgen-regtest'),'stop'])
except: rdie(1,'Unable to remove data dir!')
else:
time.sleep(2)
shutil.rmtree(data_dir)
os.mkdir(data_dir,0755)
shutil.rmtree(data_dir_enc)
os.mkdir(data_dir_enc,0755)
else:
d,pfx = '/dev/shm','mmgen-test-'
try:
@ -101,11 +103,11 @@ if not any(e in ('--skip-deps','--resume','-S','-r') for e in sys.argv+shortopts
shm_dir = tempfile.mkdtemp('',pfx,d)
except Exception as e:
die(2,'Unable to create temporary directory in {} ({})'.format(d,e))
dd = os.path.join(shm_dir,'data_dir')
dd = path_join(shm_dir,data_dir_basename,decode=False)
os.mkdir(dd,0755)
try: os.unlink(data_dir)
try: os.unlink(data_dir_enc)
except: pass
os.symlink(dd,data_dir)
os.symlink(dd,data_dir_enc)
opts_data = lambda: {
'desc': 'Test suite for the MMGen suite',
@ -500,7 +502,7 @@ cfgs = {
'ref_segwitaddrfile':'98831F3A{}-S[1,31-33,500-501,1010-1011]{}.addrs',
'ref_bech32addrfile':'98831F3A{}-B[1,31-33,500-501,1010-1011]{}.addrs',
'ref_keyaddrfile': '98831F3A{}[1,31-33,500-501,1010-1011]{}.akeys.mmenc',
'ref_passwdfile': '98831F3A-фубар@crypto.org-b58-20[1,4,9-11,1100].pws',
'ref_passwdfile': u'98831F3A-фубар@crypto.org-b58-20[1,4,9-11,1100].pws',
'ref_addrfile_chksum': {
'btc': ('6FEF 6FB9 7B13 5D91','3C2C 8558 BB54 079E'),
'ltc': ('AD52 C3FE 8924 AAF0','5738 5C4F 167C F9AE'),
@ -574,6 +576,9 @@ for a,b in (('6','11'),('7','12'),('8','13')):
cfgs[b] = deepcopy(cfgs[a])
cfgs[b]['tmpdir'] = os.path.join('test','tmp'+b)
if g.debug_utf8:
for k in cfgs: cfgs[k]['tmpdir'] += u''
from collections import OrderedDict
cmd_group = OrderedDict()
@ -1020,9 +1025,10 @@ NL = ('\r\n','\n')[g.platform=='linux' and bool(opt.popen_spawn)]
def get_file_with_ext(ext,mydir,delete=True,no_dot=False,return_list=False):
ext_enc = ext.encode('utf8')
dot = ('.','')[bool(no_dot)]
flist = [os.path.join(mydir,f) for f in os.listdir(mydir)
if f == ext or f[-len(dot+ext):] == dot+ext]
flist = [os.path.join(mydir.encode('utf8'),f).decode('utf8') for f in os.listdir(mydir.encode('utf8'))
if f == ext_enc or f[-len(dot+ext_enc):] == dot+ext_enc]
if not flist: return False
if return_list: return flist
@ -1030,7 +1036,7 @@ def get_file_with_ext(ext,mydir,delete=True,no_dot=False,return_list=False):
if len(flist) > 1:
if delete:
if not opt.quiet:
msg("Multiple *.{} files in '{}' - deleting".format(ext,mydir))
msg(u"Multiple *.{} files in '{}' - deleting".format(ext,mydir))
for f in flist:
msg(f)
os.unlink(f)
@ -1155,14 +1161,14 @@ def create_fake_unspent_data(adata,tx_data,non_mmgen_input='',non_mmgen_input_co
return out
def write_fake_data_to_file(d):
unspent_data_file = os.path.join(cfg['tmpdir'],'unspent.json')
unspent_data_file = path_join(cfg['tmpdir'],'unspent.json')
write_data_to_file(unspent_data_file,d,'Unspent outputs',silent=True)
os.environ['MMGEN_BOGUS_WALLET_DATA'] = unspent_data_file
bwd_msg = 'MMGEN_BOGUS_WALLET_DATA={}'.format(unspent_data_file)
os.environ['MMGEN_BOGUS_WALLET_DATA'] = unspent_data_file.encode('utf8')
bwd_msg = u'MMGEN_BOGUS_WALLET_DATA={}'.format(unspent_data_file)
if opt.print_cmdline: msg(bwd_msg)
if opt.log: log_fd.write(bwd_msg + ' ')
if opt.verbose or opt.exact_output:
sys.stderr.write("Fake transaction wallet data written to file '{}'\n".format(unspent_data_file))
sys.stderr.write(u"Fake transaction wallet data written to file '{}'\n".format(unspent_data_file))
def create_tx_data(sources):
tx_data,ad = {},AddrData()
@ -1210,7 +1216,7 @@ def make_txcreate_cmdline(tx_data):
def add_comments_to_addr_file(addrfile,outfile):
silence()
gmsg("Adding comments to address file '{}'".format(addrfile))
gmsg(u"Adding comments to address file '{}'".format(addrfile))
a = AddrList(addrfile)
for n,idx in enumerate(a.idxs(),1):
if n % 2: a.set_comment(idx,'Test address {}'.format(n))
@ -1331,13 +1337,13 @@ def check_deps(cmds):
def clean(usr_dirs=[]):
if opt.skip_deps: return
all_dirs = MMGenTestSuite().list_tmp_dirs()
dirs = (usr_dirs or all_dirs)
for d in sorted(dirs):
dirnums = (usr_dirs or all_dirs)
for d in sorted(dirnums):
if str(d) in all_dirs:
cleandir(all_dirs[str(d)])
else:
die(1,'{}: invalid directory number'.format(d))
cleandir(os.path.join('test','data_dir'))
cleandir(data_dir)
def skip_for_win():
if g.platform == 'win':
@ -1446,7 +1452,7 @@ class MMGenTestSuite(object):
t.expect('Enter brainwallet: ', ref_wallet_brainpass+'\n')
t.passphrase_new('new MMGen wallet',cfg['wpasswd'])
t.usr_rand(usr_rand_chars)
sid = os.path.basename(t.written_to_file('MMGen wallet').split('-')[0])
sid = os.path.basename(t.written_to_file('MMGen wallet')).split('-')[0]
refcheck('Seed ID',sid,cfg['seed_id'])
def refwalletgen(self,name): self.brainwalletgen_ref(name)
@ -1665,8 +1671,8 @@ class MMGenTestSuite(object):
t.expect('Add a comment to transaction? (y/N): ','\n')
t.expect('Save transaction? (y/N): ','y')
t.written_to_file('Transaction')
os.unlink(txfile) # our tx file replaces the original
os.system('touch ' + os.path.join(cfg['tmpdir'],'txbump'))
os.unlink(txfile.encode('utf8')) # our tx file replaces the original
os.system('touch ' + path_join(cfg['tmpdir'],'txbump',decode=False))
t.ok()
def txdo(self,name,addrfile,wallet):
@ -1757,7 +1763,7 @@ class MMGenTestSuite(object):
def export_seed(self,name,wf,desc='seed data',out_fmt='seed',pf=None):
f,t = self.walletconv_export(name,wf,desc=desc,out_fmt=out_fmt,pf=pf)
silence()
msg('{}: {}'.format(capfirst(desc),cyan(get_data_from_file(f,desc))))
msg(u'{}: {}'.format(capfirst(desc),cyan(get_data_from_file(f,desc))))
end_silence()
t.ok()
@ -1780,8 +1786,8 @@ class MMGenTestSuite(object):
# TODO: make outdir and hidden incog compatible (ignore --outdir and warn user?)
def export_incog_hidden(self,name,wf):
rf = os.path.join(cfg['tmpdir'],hincog_fn)
add_args = ['-J','{},{}'.format(rf,hincog_offset)]
rf = path_join(cfg['tmpdir'],hincog_fn)
add_args = ['-J',u'{},{}'.format(rf,hincog_offset)]
self.export_incog(
name,wf,desc='hidden incognito data',out_fmt='hi',add_args=add_args)
@ -1825,9 +1831,9 @@ class MMGenTestSuite(object):
self.addrgen_incog(name,wf,'',in_fmt='xi',desc='hex incognito data')
def addrgen_incog_hidden(self,name,wf,foo):
rf = os.path.join(cfg['tmpdir'],hincog_fn)
rf = path_join(cfg['tmpdir'],hincog_fn)
self.addrgen_incog(name,[],'',in_fmt='hi',desc='hidden incognito data',
args=['-H','{},{}'.format(rf,hincog_offset),'-l',str(hincog_seedlen)])
args=['-H',u'{},{}'.format(rf,hincog_offset),'-l',str(hincog_seedlen)])
def keyaddrgen(self,name,wf,pf=None,check_ref=False,mmtype=None):
if cfg['segwit'] and not mmtype:
@ -1864,11 +1870,11 @@ class MMGenTestSuite(object):
def ref_b32passwdgen(self,name,wf,pf):
ea = ['--base32','--passwd-len','17']
self.addrgen(name,wf,pf,check_ref=True,ftype='pass32',id_str='фубар@crypto.org',extra_args=ea)
self.addrgen(name,wf,pf,check_ref=True,ftype='pass32',id_str=u'фубар@crypto.org',extra_args=ea)
def ref_hexpasswdgen(self,name,wf,pf):
ea = ['--hex']
self.addrgen(name,wf,pf,check_ref=True,ftype='passhex',id_str='фубар@crypto.org',extra_args=ea)
self.addrgen(name,wf,pf,check_ref=True,ftype='passhex',id_str=u'фубар@crypto.org',extra_args=ea)
def txsign_keyaddr(self,name,keyaddr_file,txfile):
t = MMGenExpect(name,'mmgen-txsign', ['-d',cfg['tmpdir'],'-M',keyaddr_file,txfile])
@ -1943,10 +1949,10 @@ class MMGenTestSuite(object):
non_mm_fn = os.path.join(cfg['tmpdir'],non_mmgen_fn)
add_args = ['-d',cfg['tmpdir'],'-i','brain','-b'+cfg['bw_params'],'-p1','-k',non_mm_fn,'-M',f12]
t = self.txcreate_common(name,sources=['1','2','3','4','14'],non_mmgen_input='4',do_label=1,txdo_args=[f7,f8,f9,f10],add_args=add_args)
os.system('rm -f {}/*.sigtx'.format(cfg['tmpdir']))
os.system('rm -f {}/*.sigtx'.format(cfg['tmpdir'].encode('utf8')))
self.txsign4(name,f7,f8,f9,f10,f11,f12,txdo_handle=t)
self.txsend(name,'',txdo_handle=t)
os.system('touch ' + os.path.join(cfg['tmpdir'],'txdo'))
os.system('touch ' + path_join(cfg['tmpdir'],'txdo',decode=False))
def txbump4(self,name,f1,f2,f3,f4,f5,f6,f7,f8,f9): # f7:txfile,f9:'txdo'
non_mm_fn = os.path.join(cfg['tmpdir'],non_mmgen_fn)
@ -2090,11 +2096,11 @@ class MMGenTestSuite(object):
# Saved reference file tests
def ref_wallet_conv(self,name):
wf = os.path.join(ref_dir,cfg['ref_wallet'])
wf = path_join(ref_dir,cfg['ref_wallet'])
self.walletconv_in(name,wf,'MMGen wallet',pw=True,oo=True)
def ref_mn_conv(self,name,ext='mmwords',desc='Mnemonic data'):
wf = os.path.join(ref_dir,cfg['seed_id']+'.'+ext)
wf = path_join(ref_dir,cfg['seed_id']+'.'+ext)
self.walletconv_in(name,wf,desc,oo=True)
def ref_seed_conv(self,name):
@ -2143,8 +2149,8 @@ class MMGenTestSuite(object):
self.walletconv_out(name,'hex incognito data',out_fmt='xi',pw=True)
def ref_hincog_conv_out(self,name,extra_uopts=[]):
ic_f = os.path.join(cfg['tmpdir'],hincog_fn)
hi_parms = '{},{}'.format(ic_f,ref_wallet_incog_offset)
ic_f = path_join(cfg['tmpdir'],hincog_fn)
hi_parms = u'{},{}'.format(ic_f,ref_wallet_incog_offset)
sl_parm = '-l' + str(cfg['seed_len'])
self.walletconv_out(name,
'hidden incognito data', 'hi',
@ -2202,7 +2208,7 @@ class MMGenTestSuite(object):
def ref_addrfile_chk(self,name,ftype='addr',coin=None,subdir=None,pfx=None,mmtype=None,add_args=[]):
af_key = 'ref_{}file'.format(ftype)
af_fn = cfg[af_key].format(pfx or altcoin_pfx,'' if coin else tn_ext)
af = os.path.join(ref_dir,(subdir or ref_subdir,'')[ftype=='passwd'],af_fn)
af = path_join(ref_dir,(subdir or ref_subdir,'')[ftype=='passwd'],af_fn)
coin_arg = [] if coin == None else ['--coin='+coin]
tool_cmd = ftype.replace('segwit','').replace('bech32','')+'file_chksum'
t = MMGenExpect(name,'mmgen-tool',coin_arg+[tool_cmd,af]+add_args)
@ -2326,7 +2332,7 @@ class MMGenTestSuite(object):
def walletconv_out(self,name,desc,out_fmt='w',uopts=[],uopts_chk=[],pw=False):
opts = ['-d',cfg['tmpdir'],'-p1','-o',out_fmt] + uopts
infile = os.path.join(ref_dir,cfg['seed_id']+'.mmwords')
infile = path_join(ref_dir,cfg['seed_id']+'.mmwords')
t = MMGenExpect(name,'mmgen-walletconv',[usr_rand_arg]+opts+[infile],extra_desc='(convert)')
add_args = ['-l{}'.format(cfg['seed_len'])]
@ -2361,7 +2367,7 @@ class MMGenTestSuite(object):
def regtest_setup(self,name):
if g.testnet:
die(2,'--testnet option incompatible with regtest test suite')
try: shutil.rmtree(os.path.join(data_dir,'regtest'))
try: shutil.rmtree(os.path.join(data_dir_enc,'regtest'))
except: pass
os.environ['MMGEN_TEST_SUITE'] = '' # mnemonic is piped to stdin, so stop being a terminal
t = MMGenExpect(name,'mmgen-regtest',['-n','setup'])
@ -2383,7 +2389,7 @@ class MMGenTestSuite(object):
@staticmethod
def regtest_user_dir(user,coin=None):
return os.path.join(data_dir,'regtest',coin or g.coin.lower(),user)
return path_join(data_dir,'regtest',coin or g.coin.lower(),user)
def regtest_user_sid(self,user):
return os.path.basename(get_file_with_ext('mmdat',self.regtest_user_dir(user)))[:8]
@ -2392,7 +2398,7 @@ class MMGenTestSuite(object):
from mmgen.addr import MMGenAddrType
for mmtype in g.proto.mmtypes:
t = MMGenExpect(name,'mmgen-addrgen',
['--quiet','--'+user,'--type='+mmtype,'--outdir={}'.format(self.regtest_user_dir(user))] +
['--quiet','--'+user,'--type='+mmtype,u'--outdir={}'.format(self.regtest_user_dir(user))] +
([],[wf])[bool(wf)] + [addr_range],
extra_desc='({})'.format(MMGenAddrType.mmtypes[mmtype]['name']))
t.passphrase('MMGen wallet',passwd)
@ -2409,7 +2415,8 @@ class MMGenTestSuite(object):
for mmtype in g.proto.mmtypes:
desc = MMGenAddrType.mmtypes[mmtype]['name']
fn = os.path.join(self.regtest_user_dir(user),
'{}{}{}[{}].addrs'.format(sid,altcoin_pfx,id_strs[desc],addr_range))
u'{}{}{}[{}]{x}.addrs'.format(sid,altcoin_pfx,id_strs[desc],addr_range,
x=u'' if g.debug_utf8 else ''))
t = MMGenExpect(name,'mmgen-addrimport', ['--quiet','--'+user,'--batch',fn],extra_desc='('+desc+')')
if g.debug:
t.expect("Type uppercase 'YES' to confirm: ",'YES\n')
@ -2514,7 +2521,7 @@ class MMGenTestSuite(object):
def get_addr_from_regtest_addrlist(self,user,sid,mmtype,idx,addr_range='1-5'):
id_str = { 'L':'', 'S':'-S', 'C':'-C' }[mmtype]
ext = '{}{}{}[{}].addrs'.format(sid,altcoin_pfx,id_str,addr_range)
ext = u'{}{}{}[{}]{x}.addrs'.format(sid,altcoin_pfx,id_str,addr_range,x=u'' if g.debug_utf8 else '')
fn = get_file_with_ext(ext,self.regtest_user_dir(user),no_dot=True)
silence()
psave = g.proto
@ -2567,7 +2574,9 @@ class MMGenTestSuite(object):
t.ok()
def regtest_bob_rbf_bump(self,name):
txfile = get_file_with_ext(',{}].sigtx'.format(rtFee[1][:-1]),cfg['tmpdir'],delete=False,no_dot=True)
txfile = get_file_with_ext(u',{}]{x}.sigtx'.format(
rtFee[1][:-1],x=u'' if g.debug_utf8 else ''),
cfg['tmpdir'],delete=False,no_dot=True)
return self.regtest_user_txbump(name,'bob',txfile,rtFee[2],'c')
def regtest_generate(self,name,coin=None,num_blocks=1):