py3port: use encode() and decode() where applicable
This commit is contained in:
parent
46f44b645b
commit
04329b6757
GPG key ID:
3F8B1861E32B7DA2
28 changed files with 143 additions and 152 deletions
|
|
@ -264,7 +264,7 @@ class KeyGeneratorDummy(KeyGenerator):
|
|||
desc = 'mmgen-dummy'
|
||||
def to_pubhex(self,privhex):
|
||||
assert type(privhex) == PrivKey
|
||||
return PubKey(str(privhex),compressed=privhex.compressed)
|
||||
return PubKey(privhex.decode(),compressed=privhex.compressed)
|
||||
|
||||
class AddrListEntry(MMGenListItem):
|
||||
addr = MMGenListItemAttr('addr','CoinAddr')
|
||||
|
|
@ -473,11 +473,11 @@ Removed {{}} duplicate WIF key{{}} from keylist (also in {pnm} key-address file
|
|||
if g.proto.is_testnet():
|
||||
scramble_key += ':testnet'
|
||||
dmsg_sc('str',scramble_key)
|
||||
return scramble_seed(seed,scramble_key,self.scramble_hash_rounds)
|
||||
return scramble_seed(seed,scramble_key.encode(),self.scramble_hash_rounds)
|
||||
|
||||
def encrypt(self,desc='new key list'):
|
||||
from mmgen.crypto import mmgen_encrypt
|
||||
self.fmt_data = mmgen_encrypt(self.fmt_data.encode('utf8'),desc,'')
|
||||
self.fmt_data = mmgen_encrypt(self.fmt_data.encode(),desc,'')
|
||||
self.ext += '.'+g.mmenc_ext
|
||||
|
||||
def write_to_file(self,ask_tty=True,ask_write_default_yes=False,binary=False,desc=None):
|
||||
|
|
@ -843,7 +843,7 @@ Record this checksum: it will be used to verify the password file in the future
|
|||
def make_passwd(self,hex_sec):
|
||||
assert self.pw_fmt in self.pw_info
|
||||
if self.pw_fmt == 'hex':
|
||||
return hex_sec
|
||||
return hex_sec.decode()
|
||||
else:
|
||||
# we take least significant part
|
||||
return baseconv.fromhex(hex_sec,self.pw_fmt,pad=self.pw_len,tostr=True)[-self.pw_len:]
|
||||
|
|
@ -861,9 +861,9 @@ Record this checksum: it will be used to verify the password file in the future
|
|||
# Changing either pw_fmt, pw_len or scramble_key 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.encode('utf8'))
|
||||
scramble_key = '{}:{}:{}'.format(self.pw_fmt,self.pw_len,self.pw_id_str)
|
||||
from mmgen.crypto import scramble_seed
|
||||
return scramble_seed(seed,scramble_key,self.scramble_hash_rounds)
|
||||
return scramble_seed(seed,scramble_key.encode(),self.scramble_hash_rounds)
|
||||
|
||||
class AddrData(MMGenObject):
|
||||
msgs = {
|
||||
|
|
|
|||
|
|
@ -34,7 +34,7 @@ from mmgen.util import msg,msg_r,pmsg,pdie
|
|||
def parse_abi(s):
|
||||
return [s[:8]] + [s[8+x*64:8+(x+1)*64] for x in range(len(s[8:])//64)]
|
||||
|
||||
def create_method_id(sig): return keccak_256(sig).hexdigest()[:8]
|
||||
def create_method_id(sig): return keccak_256(sig.encode()).hexdigest()[:8]
|
||||
|
||||
class Token(MMGenObject): # ERC20
|
||||
|
||||
|
|
@ -119,7 +119,7 @@ class Token(MMGenObject): # ERC20
|
|||
return hex_tx,coin_txid
|
||||
|
||||
def txsend(self,hex_tx):
|
||||
return g.rpch.eth_sendRawTransaction('0x'+hex_tx).replace('0x','',1)
|
||||
return g.rpch.eth_sendRawTransaction('0x'+hex_tx.decode()).replace('0x','',1).encode()
|
||||
|
||||
def transfer( self,from_addr,to_addr,amt,key,start_gas,gasPrice,
|
||||
method_sig='transfer(address,uint256)',
|
||||
|
|
|
|||
|
|
@ -60,11 +60,11 @@ class EthereumMMGenTX(MMGenTX):
|
|||
|
||||
@classmethod
|
||||
def get_receipt(cls,txid):
|
||||
return g.rpch.eth_getTransactionReceipt('0x'+txid)
|
||||
return g.rpch.eth_getTransactionReceipt('0x'+txid.decode())
|
||||
|
||||
@classmethod
|
||||
def get_exec_status(cls,txid,silent=False):
|
||||
d = g.rpch.eth_getTransactionReceipt('0x'+txid)
|
||||
d = g.rpch.eth_getTransactionReceipt('0x'+txid.decode())
|
||||
if not silent:
|
||||
if 'contractAddress' in d and d['contractAddress']:
|
||||
msg('Contract address: {}'.format(d['contractAddress'].replace('0x','')))
|
||||
|
|
@ -91,6 +91,7 @@ class EthereumMMGenTX(MMGenTX):
|
|||
|
||||
# hex data if signed, json if unsigned: see create_raw()
|
||||
def check_txfile_hex_data(self):
|
||||
if type(self.hex) == str: self.hex = self.hex.encode()
|
||||
if self.check_sigs():
|
||||
from ethereum.transactions import Transaction
|
||||
import rlp
|
||||
|
|
@ -106,7 +107,7 @@ class EthereumMMGenTX(MMGenTX):
|
|||
'nonce': ETHNonce(d['nonce']),
|
||||
'data': HexStr(d['data']) }
|
||||
if o['data'] and not o['to']:
|
||||
self.token_addr = TokenAddr(etx.creates.encode('hex'))
|
||||
self.token_addr = TokenAddr(etx.creates.encode('hex')).decode()
|
||||
txid = CoinTxID(etx.hash.encode('hex'))
|
||||
assert txid == self.coin_txid,"txid in tx.hex doesn't match value in MMGen transaction file"
|
||||
else:
|
||||
|
|
@ -317,10 +318,10 @@ class EthereumMMGenTX(MMGenTX):
|
|||
|
||||
def is_in_mempool(self):
|
||||
# pmsg(g.rpch.parity_pendingTransactions())
|
||||
return '0x'+self.coin_txid in [x['hash'] for x in g.rpch.parity_pendingTransactions()]
|
||||
return '0x'+self.coin_txid.decode() in [x['hash'] for x in g.rpch.parity_pendingTransactions()]
|
||||
|
||||
def is_in_wallet(self):
|
||||
d = g.rpch.eth_getTransactionReceipt('0x'+self.coin_txid)
|
||||
d = g.rpch.eth_getTransactionReceipt('0x'+self.coin_txid.decode())
|
||||
if d and 'blockNumber' in d and d['blockNumber'] is not None:
|
||||
return 1 + int(g.rpch.eth_blockNumber(),16) - int(d['blockNumber'],16)
|
||||
return False
|
||||
|
|
@ -362,7 +363,7 @@ class EthereumMMGenTX(MMGenTX):
|
|||
|
||||
if prompt_user: self.confirm_send()
|
||||
|
||||
ret = None if bogus_send else g.rpch.eth_sendRawTransaction('0x'+self.hex,on_fail='return')
|
||||
ret = None if bogus_send else g.rpch.eth_sendRawTransaction('0x'+self.hex.decode(),on_fail='return')
|
||||
|
||||
from mmgen.rpc import rpc_error,rpc_errmsg
|
||||
if rpc_error(ret):
|
||||
|
|
@ -373,7 +374,7 @@ class EthereumMMGenTX(MMGenTX):
|
|||
else:
|
||||
m = 'BOGUS transaction NOT sent: {}' if bogus_send else 'Transaction sent: {}'
|
||||
if not bogus_send:
|
||||
assert ret == '0x'+self.coin_txid,'txid mismatch (after sending)'
|
||||
assert ret == '0x'+self.coin_txid.decode(),'txid mismatch (after sending)'
|
||||
self.desc = 'sent transaction'
|
||||
msg(m.format(self.coin_txid.hl()))
|
||||
self.add_timestamp()
|
||||
|
|
|
|||
|
|
@ -111,7 +111,7 @@ def scrypt_hash_passphrase(passwd,salt,hash_preset,buflen=32):
|
|||
# Buflen arg is for brainwallets only, which use this function to generate
|
||||
# the seed directly.
|
||||
N,r,p = get_hash_params(hash_preset)
|
||||
if type(passwd) == str: passwd = passwd.encode('utf8')
|
||||
if type(passwd) == str: passwd = passwd.encode()
|
||||
return scrypt.hash(passwd,salt,2**N,r,p,buflen=buflen)
|
||||
|
||||
def make_key(passwd,salt,hash_preset,desc='encryption key',from_what='passphrase',verbose=False):
|
||||
|
|
@ -147,7 +147,7 @@ def _get_random_data_from_user(uchars):
|
|||
prompt = 'User random data successfully acquired. Press ENTER to continue'
|
||||
prompt_and_get_char(prompt,'',enter_ok=True)
|
||||
|
||||
return key_data+''.join(fmt_time_data)
|
||||
return key_data+''.join(fmt_time_data).encode()
|
||||
|
||||
def get_random(length):
|
||||
from Crypto import Random
|
||||
|
|
|
|||
|
|
@ -100,7 +100,7 @@ class g(object):
|
|||
die(1,"'{}': platform not supported by {}\n".format(sys.platform,proj_name))
|
||||
|
||||
if os.getenv('HOME'): # Linux or MSYS
|
||||
home_dir = os.getenv('HOME').decode('utf8')
|
||||
home_dir = os.getenv('HOME')
|
||||
elif platform == 'win': # Windows native:
|
||||
die(1,'$HOME not set! {} for Windows must be run in MSYS environment'.format(proj_name))
|
||||
else:
|
||||
|
|
|
|||
|
|
@ -22,16 +22,6 @@ main.py - Script launcher for the MMGen suite
|
|||
|
||||
def launch(what):
|
||||
|
||||
def my_dec(a):
|
||||
try:
|
||||
return a.decode('utf8')
|
||||
except:
|
||||
sys.stderr.write("Argument {!r} is not a valid UTF-8 string".format(a))
|
||||
sys.exit(2)
|
||||
|
||||
import sys
|
||||
sys.argv = list(map(my_dec,sys.argv))
|
||||
|
||||
if what in ('walletgen','walletchk','walletconv','passchg'):
|
||||
what = 'wallet'
|
||||
if what == 'keygen': what = 'addrgen'
|
||||
|
|
|
|||
|
|
@ -268,7 +268,7 @@ def wipe_existing_key():
|
|||
|
||||
def create_key():
|
||||
from binascii import hexlify
|
||||
kdata = hexlify(os.urandom(32))
|
||||
kdata = hexlify(os.urandom(32)).decode()
|
||||
fn = os.path.join(tx_dir,key_fn)
|
||||
desc = 'key file {}'.format(fn)
|
||||
msg('Creating ' + desc)
|
||||
|
|
|
|||
|
|
@ -118,7 +118,7 @@ def get_data_from_cfg_file():
|
|||
|
||||
def copy_template_data(fn):
|
||||
try:
|
||||
with open(fn,'wb') as f: f.write(template_data)
|
||||
with open(fn,'wb') as f: f.write(template_data.encode())
|
||||
os.chmod(fn,0o600)
|
||||
except:
|
||||
die(2,"ERROR: unable to write to datadir '{}'".format(g.data_dir))
|
||||
|
|
@ -174,7 +174,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.decode('utf8'),getattr(g,gname),name,invert_bool))
|
||||
setattr(g,gname,set_for_type(val,getattr(g,gname),name,invert_bool))
|
||||
|
||||
def warn_altcoins(trust_level):
|
||||
if trust_level == None: return
|
||||
|
|
|
|||
|
|
@ -28,10 +28,10 @@ from mmgen.globalvars import g
|
|||
import mmgen.bech32 as bech32
|
||||
|
||||
def hash160(hexnum): # take hex, return hex - OP_HASH160
|
||||
return hashlib.new('ripemd160',hashlib.sha256(unhexlify(hexnum)).digest()).hexdigest()
|
||||
return hashlib.new('ripemd160',hashlib.sha256(unhexlify(hexnum)).digest()).hexdigest().encode()
|
||||
|
||||
def hash256(hexnum): # take hex, return hex - OP_HASH256
|
||||
return hashlib.sha256(hashlib.sha256(unhexlify(hexnum)).digest()).hexdigest()
|
||||
return hashlib.sha256(hashlib.sha256(unhexlify(hexnum)).digest()).hexdigest().encode()
|
||||
|
||||
# From en.bitcoin.it:
|
||||
# The Base58 encoding used is home made, and has some differences.
|
||||
|
|
@ -55,7 +55,7 @@ def _b58chk_encode(hexstr):
|
|||
return _numtob58(int(hexstr+hash256(hexstr)[:8],16))
|
||||
|
||||
def _b58chk_decode(s):
|
||||
hexstr = '{:x}'.format(_b58tonum(s))
|
||||
hexstr = '{:x}'.format(_b58tonum(s)).encode()
|
||||
if len(hexstr) % 2: hexstr = '0' + hexstr
|
||||
if hexstr[-8:] == hash256(hexstr[:-8])[:8]:
|
||||
return hexstr[:-8]
|
||||
|
|
@ -119,7 +119,7 @@ class BitcoinProtocol(MMGenObject):
|
|||
ydie(3,'Private key == secp256k1_ge!')
|
||||
else:
|
||||
ymsg('Warning: private key is greater than secp256k1 group order!:\n {}'.format(hexpriv))
|
||||
return '{:064x}'.format(pk % cls.secp256k1_ge)
|
||||
return '{:064x}'.format(pk % cls.secp256k1_ge).encode()
|
||||
|
||||
@classmethod
|
||||
def hex2wif(cls,hexpriv,pubkey_type,compressed):
|
||||
|
|
@ -167,7 +167,7 @@ class BitcoinProtocol(MMGenObject):
|
|||
if num == False:
|
||||
if g.debug: Msg('Address cannot be converted to base 58')
|
||||
break
|
||||
addr_hex = '{:0{}x}'.format(num,len(ver_num)+hex_width+8)
|
||||
addr_hex = '{:0{}x}'.format(num,len(ver_num)+hex_width+8).encode()
|
||||
# pmsg(hex_width,len(addr_hex),addr_hex[:len(ver_num)],ver_num)
|
||||
if addr_hex[:len(ver_num)] != ver_num: continue
|
||||
if hash256(addr_hex[:-8])[:8] == addr_hex[-8:]:
|
||||
|
|
@ -297,11 +297,11 @@ class DummyWIF(object):
|
|||
n = cls.name.capitalize()
|
||||
assert pubkey_type == cls.pubkey_type,'{}: invalid pubkey_type for {}!'.format(pubkey_type,n)
|
||||
assert compressed == False,'{} does not support compressed pubkeys!'.format(n)
|
||||
return str(hexpriv)
|
||||
return hexpriv.decode()
|
||||
|
||||
@classmethod
|
||||
def wif2hex(cls,wif):
|
||||
return { 'hex':str(wif), 'pubkey_type':cls.pubkey_type, 'compressed':False }
|
||||
return { 'hex':wif.encode(), 'pubkey_type':cls.pubkey_type, 'compressed':False }
|
||||
|
||||
class EthereumProtocol(DummyWIF,BitcoinProtocol):
|
||||
|
||||
|
|
@ -327,7 +327,7 @@ class EthereumProtocol(DummyWIF,BitcoinProtocol):
|
|||
def verify_addr(cls,addr,hex_width,return_dict=False):
|
||||
from mmgen.util import is_hex_str_lc
|
||||
if is_hex_str_lc(addr) and len(addr) == cls.addr_width:
|
||||
return { 'hex': addr, 'format': 'ethereum' } if return_dict else True
|
||||
return { 'hex': addr.encode(), 'format': 'ethereum' } if return_dict else True
|
||||
if g.debug: Msg("Invalid address '{}'".format(addr))
|
||||
return False
|
||||
|
||||
|
|
@ -335,7 +335,7 @@ class EthereumProtocol(DummyWIF,BitcoinProtocol):
|
|||
def pubhash2addr(cls,pubkey_hash,p2sh):
|
||||
assert len(pubkey_hash) == 40,'{}: invalid length for pubkey hash'.format(len(pubkey_hash))
|
||||
assert not p2sh,'Ethereum has no P2SH address format'
|
||||
return pubkey_hash
|
||||
return pubkey_hash.decode()
|
||||
|
||||
class EthereumTestnetProtocol(EthereumProtocol):
|
||||
data_subdir = 'testnet'
|
||||
|
|
@ -367,7 +367,7 @@ class ZcashProtocol(BitcoinProtocolAddrgen):
|
|||
@classmethod
|
||||
def preprocess_key(cls,hexpriv,pubkey_type): # zero the first four bits
|
||||
if pubkey_type == 'zcash_z':
|
||||
return '{:02x}'.format(int(hexpriv[:2],16) & 0x0f) + hexpriv[2:]
|
||||
return '{:02x}'.format(int(hexpriv[:2],16) & 0x0f).encode() + hexpriv[2:]
|
||||
else:
|
||||
return hexpriv
|
||||
|
||||
|
|
@ -428,7 +428,7 @@ class MoneroProtocol(DummyWIF,BitcoinProtocolAddrgen):
|
|||
chk = sha3.keccak_256(ret.decode('hex')[:-4]).hexdigest()[:8]
|
||||
assert chk == ret[-8:],'Incorrect checksum'
|
||||
|
||||
return { 'hex': ret, 'format': 'monero' } if return_dict else True
|
||||
return { 'hex': ret.encode(), 'format': 'monero' } if return_dict else True
|
||||
|
||||
class MoneroTestnetProtocol(MoneroProtocol):
|
||||
addr_ver_num = { 'monero': (b'35','4'), 'monero_sub': (b'3f','8') } # 53,63
|
||||
|
|
|
|||
|
|
@ -162,7 +162,7 @@ def show_mempool():
|
|||
p = start_cmd('cli','getrawmempool')
|
||||
from pprint import pformat
|
||||
from ast import literal_eval
|
||||
msg(pformat(literal_eval(p.stdout.read())))
|
||||
msg(pformat(literal_eval(p.stdout.read().decode())))
|
||||
p.wait()
|
||||
|
||||
def cli(*args):
|
||||
|
|
@ -246,7 +246,7 @@ def get_current_user_win(quiet=False):
|
|||
|
||||
def get_current_user_unix(quiet=False):
|
||||
p = start_cmd('pgrep','-af','{}.*--rpcport={}.*'.format(g.proto.daemon_name,rpc_port),quiet=True)
|
||||
cmdline = p.stdout.read()
|
||||
cmdline = p.stdout.read().decode()
|
||||
if not cmdline: return None
|
||||
for k in ('miner','bob','alice'):
|
||||
if 'wallet.dat.'+k in cmdline:
|
||||
|
|
@ -296,7 +296,7 @@ def generate(blocks=1,silent=False):
|
|||
p = start_cmd('cli','generate',str(blocks))
|
||||
out = process_output(p,silent=silent)[0]
|
||||
from ast import literal_eval
|
||||
if len(literal_eval(out)) != blocks:
|
||||
if len(literal_eval(out.decode())) != blocks:
|
||||
rdie(1,'Error generating blocks')
|
||||
p.wait()
|
||||
gmsg('Mined {} block{}'.format(blocks,suf(blocks,'s')))
|
||||
|
|
|
|||
|
|
@ -119,9 +119,9 @@ class CoinDaemonRPCConnection(object):
|
|||
http_hdr = { 'Content-Type': 'application/json' }
|
||||
if self.auth:
|
||||
fs = ' RPC AUTHORIZATION data ==> raw: [{}]\n{:>31}enc: [Basic {}]\n'
|
||||
as_enc = base64.b64encode(self.auth_str)
|
||||
as_enc = base64.b64encode(self.auth_str.encode())
|
||||
dmsg_rpc(fs.format(self.auth_str,'',as_enc))
|
||||
http_hdr.update({ 'Host':self.host, 'Authorization':'Basic {}'.format(as_enc) })
|
||||
http_hdr.update({ 'Host':self.host, 'Authorization':'Basic {}'.format(as_enc.decode()) })
|
||||
|
||||
try:
|
||||
hc.request('POST','/',json.dumps(p,cls=MyJSONEncoder),http_hdr)
|
||||
|
|
@ -149,7 +149,7 @@ class CoinDaemonRPCConnection(object):
|
|||
e2 = str(e1)
|
||||
return do_fail(r,1,e2)
|
||||
|
||||
r2 = r.read().decode('utf8')
|
||||
r2 = r.read()
|
||||
|
||||
dmsg_rpc(' RPC REPLY data ==> {}\n'.format(r2))
|
||||
|
||||
|
|
|
|||
|
|
@ -371,7 +371,7 @@ class Mnemonic (SeedSourceUnenc):
|
|||
prompt = 'Choose a mnemonic length: 1) 12 words, 2) 18 words, 3) 24 words: '
|
||||
urange = [str(i+1) for i in range(len(self.mn_lens))]
|
||||
while True:
|
||||
r = get_char('\r'+prompt)
|
||||
r = get_char('\r'+prompt).decode()
|
||||
if r in urange: break
|
||||
msg_r('\r' + ' '*len(prompt) + '\r')
|
||||
return self.mn_lens[int(r)-1]
|
||||
|
|
@ -396,7 +396,7 @@ class Mnemonic (SeedSourceUnenc):
|
|||
def get_word():
|
||||
s,pad = '',0
|
||||
while True:
|
||||
ch = get_char_raw('')
|
||||
ch = get_char_raw('').decode()
|
||||
if ch in '\b\x7f':
|
||||
if s: s = s[:-1]
|
||||
elif ch in '\n ':
|
||||
|
|
@ -439,7 +439,7 @@ class Mnemonic (SeedSourceUnenc):
|
|||
hexseed = self.seed.hexdata
|
||||
|
||||
mn = baseconv.fromhex(hexseed,self.wl_id,self._hex2mn_pad(hexseed))
|
||||
ret = baseconv.tohex(mn,self.wl_id,self._mn2hex_pad(mn))
|
||||
ret = baseconv.tohex(mn,self.wl_id,self._mn2hex_pad(mn)).encode()
|
||||
|
||||
# Internal error, so just die on fail
|
||||
compare_or_die(ret,'recomputed seed',hexseed,'original',e='Internal error')
|
||||
|
|
@ -461,7 +461,7 @@ class Mnemonic (SeedSourceUnenc):
|
|||
msg('Invalid mnemonic: word #{} is not in the wordlist'.format(n))
|
||||
return False
|
||||
|
||||
hexseed = baseconv.tohex(mn,self.wl_id,self._mn2hex_pad(mn))
|
||||
hexseed = baseconv.tohex(mn,self.wl_id,self._mn2hex_pad(mn)).encode()
|
||||
ret = baseconv.fromhex(hexseed,self.wl_id,self._hex2mn_pad(hexseed))
|
||||
|
||||
if len(hexseed) * 4 not in g.seed_lens:
|
||||
|
|
@ -539,7 +539,7 @@ class HexSeedFile (SeedSourceUnenc):
|
|||
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.decode()))
|
||||
|
||||
def _deformat(self):
|
||||
desc = self.desc
|
||||
|
|
@ -642,7 +642,7 @@ class Wallet (SeedSourceEnc):
|
|||
'{} {}'.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('utf8'))
|
||||
chksum = make_chksum_6(' '.join(lines).encode())
|
||||
self.fmt_data = '\n'.join((chksum,)+lines) + '\n'
|
||||
|
||||
def _deformat(self):
|
||||
|
|
|
|||
|
|
@ -44,7 +44,7 @@ def print_help_and_exit(opts_data,longhelp=False):
|
|||
n = opts_data['notes']
|
||||
if isinstance(n, collections.Callable): n = n()
|
||||
out += '\n ' + '\n '.join(n.rstrip().splitlines())
|
||||
print((out.encode('utf8')))
|
||||
print(out)
|
||||
sys.exit(0)
|
||||
|
||||
def process_opts(argv,opts_data,short_opts,long_opts,skip_help=False):
|
||||
|
|
|
|||
|
|
@ -85,7 +85,9 @@ def _get_keypress_unix_raw(prompt='',immed_chars='',prehold_protect=None):
|
|||
|
||||
def _get_keypress_unix_stub(prompt='',immed_chars='',prehold_protect=None):
|
||||
msg_r(prompt)
|
||||
return sys.stdin.read(1)
|
||||
return sys.stdin.read(1).encode()
|
||||
|
||||
#_get_keypress_unix_stub = _get_keypress_unix
|
||||
|
||||
def _kb_hold_protect_mswin():
|
||||
|
||||
|
|
|
|||
|
|
@ -27,7 +27,7 @@ from mmgen.common import *
|
|||
|
||||
# Windows uses non-UTF8 encodings in filesystem, so use raw bytes here
|
||||
def cleandir(d):
|
||||
d_enc = d.encode('utf8')
|
||||
d_enc = d.encode()
|
||||
|
||||
try: files = os.listdir(d_enc)
|
||||
except: return
|
||||
|
|
@ -41,7 +41,7 @@ def cleandir(d):
|
|||
rmtree(os.path.join(d_enc,f))
|
||||
|
||||
def getrandnum(n): return int(hexlify(os.urandom(n)),16)
|
||||
def getrandhex(n): return hexlify(os.urandom(n))
|
||||
def getrandhex(n): return hexlify(os.urandom(n)).decode()
|
||||
def getrandnum_range(nbytes,rn_max):
|
||||
while True:
|
||||
rn = int(hexlify(os.urandom(nbytes)),16)
|
||||
|
|
|
|||
|
|
@ -225,10 +225,10 @@ def are_equal(a,b,dtype=''):
|
|||
def print_convert_results(indata,enc,dec,dtype):
|
||||
error = (True,False)[are_equal(indata,dec,dtype)]
|
||||
if error or opt.verbose:
|
||||
Msg('Input: {}'.format(repr(indata)))
|
||||
Msg('Encoded data: {}'.format(repr(enc)))
|
||||
Msg('Recoded data: {}'.format(repr(dec)))
|
||||
else: Msg(enc)
|
||||
Msg('Input: {}'.format(indata))
|
||||
Msg('Encoded data: {}'.format(enc))
|
||||
Msg('Recoded data: {}'.format(dec))
|
||||
else: Msg(enc.decode())
|
||||
if error:
|
||||
die(3,"Error! Recoded data doesn't match input!")
|
||||
|
||||
|
|
@ -253,10 +253,10 @@ def B58randenc():
|
|||
r = get_random(32)
|
||||
enc = baseconv.b58encode(r,pad=True)
|
||||
dec = baseconv.b58decode(enc,pad=True)
|
||||
print_convert_results(r,enc,dec,'str')
|
||||
print_convert_results(r,enc.encode(),dec,'bytes')
|
||||
|
||||
def Randhex(nbytes='32'):
|
||||
Msg(binascii.hexlify(get_random(int(nbytes))))
|
||||
Msg(binascii.hexlify(get_random(int(nbytes))).decode())
|
||||
|
||||
def Randwif():
|
||||
Msg(PrivKey(get_random(32),pubkey_type=at.pubkey_type,compressed=at.compressed).wif)
|
||||
|
|
@ -277,37 +277,37 @@ def Wif2segwit_pair(wif):
|
|||
pubhex = kg.to_pubhex(PrivKey(wif=wif))
|
||||
addr = ag.to_addr(pubhex)
|
||||
rs = ag.to_segwit_redeem_script(pubhex)
|
||||
Msg('{}\n{}'.format(rs,addr))
|
||||
Msg('{}\n{}'.format(rs.decode(),addr))
|
||||
|
||||
def Pubhash2addr(pubhash):
|
||||
if opt.type == 'bech32':
|
||||
ret = g.proto.pubhash2bech32addr(pubhash)
|
||||
ret = g.proto.pubhash2bech32addr(pubhash.encode())
|
||||
else:
|
||||
ret = g.proto.pubhash2addr(pubhash,at.addr_fmt=='p2sh')
|
||||
ret = g.proto.pubhash2addr(pubhash.encode(),at.addr_fmt=='p2sh')
|
||||
Msg(ret)
|
||||
|
||||
def Addr2hexaddr(addr): Msg(g.proto.verify_addr(addr,CoinAddr.hex_width,return_dict=True)['hex'])
|
||||
def Hash160(pubkeyhex): Msg(hash160(pubkeyhex))
|
||||
def Pubhex2addr(pubkeyhex): Pubhash2addr(hash160(pubkeyhex))
|
||||
def Wif2hex(wif): Msg(PrivKey(wif=wif))
|
||||
def Addr2hexaddr(addr): Msg(g.proto.verify_addr(addr,CoinAddr.hex_width,return_dict=True)['hex'].decode())
|
||||
def Hash160(pubkeyhex): Msg(hash160(pubkeyhex).decode())
|
||||
def Pubhex2addr(pubkeyhex): Pubhash2addr(hash160(pubkeyhex.encode()).decode())
|
||||
def Wif2hex(wif): Msg(PrivKey(wif=wif).decode())
|
||||
|
||||
def Hex2wif(hexpriv):
|
||||
Msg(g.proto.hex2wif(hexpriv,pubkey_type=at.pubkey_type,compressed=at.compressed))
|
||||
Msg(g.proto.hex2wif(hexpriv.encode(),pubkey_type=at.pubkey_type,compressed=at.compressed))
|
||||
|
||||
def Privhex2addr(privhex,output_pubhex=False):
|
||||
pk = PrivKey(binascii.unhexlify(privhex),compressed=at.compressed,pubkey_type=at.pubkey_type)
|
||||
ph = kg.to_pubhex(pk)
|
||||
Msg(ph if output_pubhex else ag.to_addr(ph))
|
||||
Msg(ph.decode() if output_pubhex else ag.to_addr(ph))
|
||||
|
||||
def Privhex2pubhex(privhex): # new
|
||||
Privhex2addr(privhex,output_pubhex=True)
|
||||
|
||||
def Pubhex2redeem_script(pubhex): # new
|
||||
Msg(g.proto.pubhex2redeem_script(pubhex))
|
||||
Msg(g.proto.pubhex2redeem_script(pubhex).decode())
|
||||
|
||||
def Wif2redeem_script(wif): # new
|
||||
privhex = PrivKey(wif=wif)
|
||||
Msg(ag.to_segwit_redeem_script(kg.to_pubhex(privhex)))
|
||||
Msg(ag.to_segwit_redeem_script(kg.to_pubhex(privhex)).decode())
|
||||
|
||||
wordlists = 'electrum','tirosh'
|
||||
dfl_wl_id = 'electrum'
|
||||
|
|
@ -325,16 +325,16 @@ def Mn_rand128(wordlist=dfl_wl_id): do_random_mn(16,wordlist)
|
|||
def Mn_rand192(wordlist=dfl_wl_id): do_random_mn(24,wordlist)
|
||||
def Mn_rand256(wordlist=dfl_wl_id): do_random_mn(32,wordlist)
|
||||
|
||||
def Hex2mn(s,wordlist=dfl_wl_id): Msg(' '.join(baseconv.fromhex(s,wordlist)))
|
||||
def Hex2mn(s,wordlist=dfl_wl_id): Msg(' '.join(baseconv.fromhex(s.encode(),wordlist)))
|
||||
def Mn2hex(s,wordlist=dfl_wl_id): Msg(baseconv.tohex(s.split(),wordlist))
|
||||
|
||||
def Strtob58(s,pad=None): Msg(baseconv.fromhex(binascii.hexlify(s),'b58',pad,tostr=True))
|
||||
def Hextob58(s,pad=None): Msg(baseconv.fromhex(s,'b58',pad,tostr=True))
|
||||
def Strtob58(s,pad=None): Msg(baseconv.fromhex(binascii.hexlify(s.encode()),'b58',pad,tostr=True))
|
||||
def Hextob58(s,pad=None): Msg(baseconv.fromhex(s.encode(),'b58',pad,tostr=True))
|
||||
def Hextob58chk(s):
|
||||
from mmgen.protocol import _b58chk_encode
|
||||
Msg(_b58chk_encode(s))
|
||||
def Hextob32(s,pad=None): Msg(baseconv.fromhex(s,'b32',pad,tostr=True))
|
||||
def B58tostr(s): Msg(binascii.unhexlify(baseconv.tohex(s,'b58')))
|
||||
def Hextob32(s,pad=None): Msg(baseconv.fromhex(s.encode(),'b32',pad,tostr=True))
|
||||
def B58tostr(s): Msg(binascii.unhexlify(baseconv.tohex(s,'b58')).decode())
|
||||
def B58tohex(s,pad=None): Msg(baseconv.tohex(s,'b58',pad))
|
||||
def B58chktohex(s):
|
||||
from mmgen.protocol import _b58chk_decode
|
||||
|
|
@ -375,17 +375,17 @@ def Passwdfile_chksum(infile):
|
|||
PasswordList(infile=infile,chksum_only=True)
|
||||
|
||||
def Hexreverse(s):
|
||||
Msg(binascii.hexlify(binascii.unhexlify(s.strip())[::-1]))
|
||||
Msg(binascii.hexlify(binascii.unhexlify(s.strip())[::-1]).decode())
|
||||
|
||||
def Hexlify(s):
|
||||
Msg(binascii.hexlify(s))
|
||||
Msg(binascii.hexlify(s.encode()).decode())
|
||||
|
||||
def Hash256(s,file_input=False,hex_input=False):
|
||||
from hashlib import sha256
|
||||
if file_input: b = get_data_from_file(s,binary=True)
|
||||
elif hex_input: b = decode_pretty_hexdump(s)
|
||||
else: b = s
|
||||
Msg(sha256(sha256(b).digest()).hexdigest())
|
||||
Msg(sha256(sha256(b.encode()).digest()).hexdigest())
|
||||
|
||||
def Encrypt(infile,outfile='',hash_preset=''):
|
||||
data = get_data_from_file(infile,'data for encryption',binary=True)
|
||||
|
|
@ -512,7 +512,7 @@ def monero_wallet_ops(infile,op,blockheight=None,addrs=None):
|
|||
die(1,"Unable to run 'monero-wallet-cli'!")
|
||||
p = run_cmd(['monerod','status'])
|
||||
import re
|
||||
m = re.search(r'Height: (\d+)/\d+ ',p.stdout.read())
|
||||
m = re.search(r'Height: (\d+)/\d+ ',p.stdout.read().decode())
|
||||
if not m:
|
||||
die(1,'Unable to connect to monerod!')
|
||||
return int(m.group(1))
|
||||
|
|
@ -538,10 +538,10 @@ def monero_wallet_ops(infile,op,blockheight=None,addrs=None):
|
|||
try: os.stat(fn)
|
||||
except: pass
|
||||
else: die(1,"Wallet '{}' already exists!".format(fn))
|
||||
p = pexpect.spawn('monero-wallet-cli --generate-from-spend-key {}'.format(fn.encode('utf8')))
|
||||
p = pexpect.spawn('monero-wallet-cli --generate-from-spend-key {}'.format(fn))
|
||||
if g.debug: p.logfile = sys.stdout
|
||||
my_expect(p,'Awaiting initial prompt','Secret spend key: ')
|
||||
my_sendline(p,'',d.sec,65)
|
||||
my_sendline(p,'',d.sec.decode(),65)
|
||||
my_expect(p,'','Enter.* new.* password.*: ',regex=True)
|
||||
my_sendline(p,'Sending password',d.wallet_passwd,33)
|
||||
my_expect(p,'','Confirm password: ')
|
||||
|
|
@ -550,12 +550,12 @@ def monero_wallet_ops(infile,op,blockheight=None,addrs=None):
|
|||
my_sendline(p,'','1',2)
|
||||
my_expect(p,'monerod generating wallet','Generated new wallet: ')
|
||||
my_expect(p,'','\n')
|
||||
if d.addr not in p.before:
|
||||
die(3,'Addresses do not match!\n MMGen: {}\n Monero: {}'.format(d.addr,p.before))
|
||||
if d.addr not in p.before.decode():
|
||||
die(3,'Addresses do not match!\n MMGen: {}\n Monero: {}'.format(d.addr,p.before.decode()))
|
||||
my_expect(p,'','View key: ')
|
||||
my_expect(p,'','\n')
|
||||
if d.viewkey not in p.before:
|
||||
die(3,'View keys do not match!\n MMGen: {}\n Monero: {}'.format(d.viewkey,p.before))
|
||||
if d.viewkey not in p.before.decode():
|
||||
die(3,'View keys do not match!\n MMGen: {}\n Monero: {}'.format(d.viewkey,p.before.decode()))
|
||||
my_expect(p,'','(YYYY-MM-DD): ')
|
||||
h = str(blockheight or cur_height-1)
|
||||
my_sendline(p,'',h,len(h)+1)
|
||||
|
|
@ -574,7 +574,7 @@ def monero_wallet_ops(infile,op,blockheight=None,addrs=None):
|
|||
def sync(n,d,fn):
|
||||
try: os.stat(fn)
|
||||
except: die(1,"Wallet '{}' does not exist!".format(fn))
|
||||
p = pexpect.spawn('monero-wallet-cli --wallet-file={}'.format(fn.encode('utf8')))
|
||||
p = pexpect.spawn('monero-wallet-cli --wallet-file={}'.format(fn))
|
||||
if g.debug: p.logfile = sys.stdout
|
||||
my_expect(p,'Awaiting password prompt','Wallet password: ')
|
||||
my_sendline(p,'Sending password',d.wallet_passwd,33)
|
||||
|
|
@ -592,7 +592,7 @@ def monero_wallet_ops(infile,op,blockheight=None,addrs=None):
|
|||
msg('\r Block {h} / {h}'.format(h=height))
|
||||
else:
|
||||
msg(' Wallet in sync')
|
||||
b = [l for l in p.before.splitlines() if l[:8] == 'Balance:'][0].split()
|
||||
b = [l for l in p.before.decode().splitlines() if len(l) > 7 and l[:8] == 'Balance:'][0].split()
|
||||
msg(' Balance: {} Unlocked balance: {}'.format(b[1],b[4]))
|
||||
bals[fn] = ( Decimal(b[1][:-1]), Decimal(b[4]) )
|
||||
my_sendline(p,'Exiting','exit',5)
|
||||
|
|
|
|||
|
|
@ -220,7 +220,7 @@ watch-only wallet using '{}-addrimport' and then re-run this program.
|
|||
out.append(fs.format( n=str(n+1)+')',
|
||||
t='' if not i.txid else \
|
||||
' ' * (tx_w-4) + '|...' if i.skip == 'txid' \
|
||||
else i.txid[:tx_w-len(txdots)]+txdots,
|
||||
else i.txid.decode()[:tx_w-len(txdots)]+txdots,
|
||||
v=i.vout,
|
||||
a=addr_out,
|
||||
A=i.amt.fmt(color=True,prec=self.disp_prec),
|
||||
|
|
@ -315,7 +315,7 @@ watch-only wallet using '{}-addrimport' and then re-run this program.
|
|||
while True:
|
||||
msg_r('' if no_output else '\n\n' if opt.no_blank else CUR_HOME+ERASE_ALL)
|
||||
reply = get_char('' if no_output else self.format_for_display()+'\n'+(oneshot_msg or '')+prompt,
|
||||
immed_chars=self.key_mappings)
|
||||
immed_chars=''.join(self.key_mappings.keys())).decode()
|
||||
no_output = False
|
||||
oneshot_msg = '' if oneshot_msg else None # tristate, saves previous state
|
||||
if reply not in self.key_mappings:
|
||||
|
|
|
|||
14
mmgen/tx.py
14
mmgen/tx.py
|
|
@ -661,15 +661,15 @@ Selected non-{pnm} inputs: {{}}""".strip().format(pnm=g.proj_name,pnl=g.proj_nam
|
|||
self.blockcount,
|
||||
('',' LT={}'.format(self.locktime))[bool(self.locktime)]
|
||||
),
|
||||
self.hex,
|
||||
self.hex.decode(),
|
||||
repr([amt_to_str(e.__dict__) for e in self.inputs]),
|
||||
repr([amt_to_str(e.__dict__) for e in self.outputs])
|
||||
]
|
||||
if self.label:
|
||||
lines.append(baseconv.b58encode(self.label.encode('utf8')))
|
||||
lines.append(baseconv.b58encode(self.label.encode()))
|
||||
if self.coin_txid:
|
||||
if not self.label: lines.append('-') # keep old tx files backwards compatible
|
||||
lines.append(self.coin_txid)
|
||||
lines.append(self.coin_txid.decode())
|
||||
self.chksum = make_chksum_6(' '.join(lines))
|
||||
self.fmt_data = '\n'.join([self.chksum] + lines)+'\n'
|
||||
|
||||
|
|
@ -735,7 +735,7 @@ Selected non-{pnm} inputs: {{}}""".strip().format(pnm=g.proj_name,pnl=g.proj_nam
|
|||
self.check_hex_tx_matches_mmgen_tx(dt)
|
||||
self.coin_txid = CoinTxID(dt['txid'],on_fail='raise')
|
||||
self.check_sigs(dt)
|
||||
if not self.coin_txid == g.rpch.decoderawtransaction(self.hex)['txid']:
|
||||
if not self.coin_txid.decode() == g.rpch.decoderawtransaction(ret['hex'])['txid']:
|
||||
raise BadMMGenTxID('txid mismatch (after signing)')
|
||||
msg('OK')
|
||||
return True
|
||||
|
|
@ -941,7 +941,7 @@ Selected non-{pnm} inputs: {{}}""".strip().format(pnm=g.proj_name,pnl=g.proj_nam
|
|||
if bogus_send:
|
||||
m = 'BOGUS transaction NOT sent: {}'
|
||||
else:
|
||||
assert ret == self.coin_txid, 'txid mismatch (after sending)'
|
||||
assert ret.encode() == self.coin_txid, 'txid mismatch (after sending)'
|
||||
m = 'Transaction sent: {}'
|
||||
self.desc = 'sent transaction'
|
||||
msg(m.format(self.coin_txid.hl()))
|
||||
|
|
@ -1039,7 +1039,7 @@ Selected non-{pnm} inputs: {{}}""".strip().format(pnm=g.proj_name,pnl=g.proj_nam
|
|||
((n+1,'')[ip],'address:',e.addr.fmt(color=True,width=addr_w) + ' '+mmid_fmt),
|
||||
('','comment:',e.label.hl() if e.label else ''),
|
||||
('','amount:','{} {}'.format(e.amt.hl(),g.dcoin))]
|
||||
items = [(n+1, 'tx,vout:','{},{}'.format(e.txid,e.vout))] + icommon + [
|
||||
items = [(n+1, 'tx,vout:','{},{}'.format(e.txid.decode(),e.vout))] + icommon + [
|
||||
('','confirmations:','{} (around {} days)'.format(confs,days) if blockcount else '')
|
||||
] if ip else icommon + [
|
||||
('','change:',green('True') if e.is_chg else '')]
|
||||
|
|
@ -1142,7 +1142,7 @@ Selected non-{pnm} inputs: {{}}""".strip().format(pnm=g.proj_name,pnl=g.proj_nam
|
|||
desc = 'data'
|
||||
assert len(tx_data) <= g.max_tx_file_size,(
|
||||
'Transaction file size exceeds limit ({} bytes)'.format(g.max_tx_file_size))
|
||||
tx_data = tx_data.decode('ascii').splitlines()
|
||||
tx_data = tx_data.splitlines()
|
||||
assert len(tx_data) >= 5,'number of lines less than 5'
|
||||
assert len(tx_data[0]) == 6,'invalid length of first line'
|
||||
self.chksum = HexStr(tx_data.pop(0),on_fail='raise')
|
||||
|
|
|
|||
|
|
@ -27,10 +27,10 @@ from string import hexdigits
|
|||
from mmgen.color import *
|
||||
from mmgen.exception import *
|
||||
|
||||
def msg(s): sys.stderr.write(s.encode('utf8') + '\n')
|
||||
def msg_r(s): sys.stderr.write(s.encode('utf8'))
|
||||
def Msg(s): sys.stdout.write(s.encode('utf8') + '\n')
|
||||
def Msg_r(s): sys.stdout.write(s.encode('utf8'))
|
||||
def msg(s): sys.stderr.write(s.encode() + b'\n')
|
||||
def msg_r(s): sys.stderr.write(s.encode())
|
||||
def Msg(s): sys.stdout.write(s.encode() + b'\n')
|
||||
def Msg_r(s): sys.stdout.write(s.encode())
|
||||
def msgred(s): msg(red(s))
|
||||
def rmsg(s): msg(red(s))
|
||||
def rmsg_r(s): msg_r(red(s))
|
||||
|
|
@ -167,6 +167,7 @@ def remove_extension(f,e):
|
|||
return (f,a)[len(b)>1 and b[1:]==e]
|
||||
|
||||
def make_chksum_N(s,nchars,sep=False):
|
||||
if type(s) == str: s = s.encode()
|
||||
if nchars%4 or not (4 <= nchars <= 64): return False
|
||||
s = sha256(sha256(s).digest()).hexdigest().upper()
|
||||
sep = ('',' ')[bool(sep)]
|
||||
|
|
@ -178,7 +179,7 @@ def make_chksum_8(s,sep=False):
|
|||
return '{} {}'.format(s[:4],s[4:]) if sep else s
|
||||
def make_chksum_6(s):
|
||||
from mmgen.obj import HexStr
|
||||
if type(s) == str: s = s.encode('utf8')
|
||||
if type(s) == str: s = s.encode()
|
||||
return HexStr(sha256(s).hexdigest()[:6])
|
||||
def is_chksum_6(s): return len(s) == 6 and is_hex_str_lc(s)
|
||||
|
||||
|
|
@ -296,7 +297,7 @@ class baseconv(object):
|
|||
|
||||
@classmethod
|
||||
def get_wordlist_chksum(cls,wl_id):
|
||||
return sha256(' '.join(cls.digits[wl_id])).hexdigest()[:8]
|
||||
return sha256(' '.join(cls.digits[wl_id]).encode()).hexdigest()[:8]
|
||||
|
||||
@classmethod
|
||||
def check_wordlists(cls):
|
||||
|
|
@ -334,9 +335,8 @@ class baseconv(object):
|
|||
@classmethod
|
||||
def fromhex(cls,hexnum,wl_id,pad=None,tostr=False):
|
||||
|
||||
hexnum = hexnum.strip()
|
||||
if not is_hex_str(hexnum):
|
||||
die(2,"'{}': not a hexadecimal number".format(hexnum))
|
||||
if not is_hex_str(hexnum.decode()):
|
||||
die(2,"'{}': not a hexadecimal number".format(hexnum.decode()))
|
||||
|
||||
wl = cls.digits[wl_id]
|
||||
base = len(wl)
|
||||
|
|
@ -377,7 +377,7 @@ def pretty_hexdump(data,gw=2,cols=8,line_nums=False):
|
|||
return ''.join(
|
||||
[
|
||||
('' if (line_nums == False or i % cols) else '{:06x}: '.format(i*gw)) +
|
||||
hexlify(data[i*gw:i*gw+gw]) + ('\n',' ')[bool((i+1) % cols)]
|
||||
hexlify(data[i*gw:i*gw+gw]).decode() + ('\n',' ')[bool((i+1) % cols)]
|
||||
for i in range(len(data)//gw + r)
|
||||
]
|
||||
).rstrip() + '\n'
|
||||
|
|
@ -547,9 +547,6 @@ def write_data_to_file( outfile,data,desc='data',
|
|||
if ask_write_default_yes == False or ask_write_prompt:
|
||||
ask_write = True
|
||||
|
||||
if not binary and type(data) == str:
|
||||
data = data.encode('utf8')
|
||||
|
||||
def do_stdout():
|
||||
qmsg('Output to STDOUT requested')
|
||||
if sys.stdout.isatty():
|
||||
|
|
@ -577,7 +574,7 @@ def write_data_to_file( outfile,data,desc='data',
|
|||
import msvcrt
|
||||
msvcrt.setmode(sys.stdout.fileno(),os.O_BINARY)
|
||||
|
||||
sys.stdout.write(data)
|
||||
sys.stdout.write(data.decode() if issubclass(type(data),bytes) else data)
|
||||
|
||||
def do_file(outfile,ask_write_prompt):
|
||||
if opt.outdir and not ignore_opt_outdir and not os.path.isabs(outfile):
|
||||
|
|
@ -637,7 +634,7 @@ def get_words_from_file(infile,desc,silent=False):
|
|||
if not silent:
|
||||
qmsg("Getting {} from file '{}'".format(desc,infile))
|
||||
f = open_file_or_exit(infile, 'r')
|
||||
try: words = f.read().decode('utf8').split() # split() also strips
|
||||
try: words = f.read().split() # split() also strips
|
||||
except: die(1,'{} data must be UTF-8 encoded.'.format(capfirst(desc)))
|
||||
f.close()
|
||||
dmsg('Sanitized input: [{}]'.format(' '.join(words)))
|
||||
|
|
@ -680,7 +677,7 @@ def get_data_from_file(infile,desc='data',dash=False,silent=False,binary=False,r
|
|||
data = f.read()
|
||||
f.close()
|
||||
if require_utf8:
|
||||
try: data = data.decode('utf8')
|
||||
try: data = data.decode()
|
||||
except: die(1,'{} data must be UTF-8 encoded.'.format(capfirst(desc)))
|
||||
return data
|
||||
|
||||
|
|
@ -714,14 +711,14 @@ def my_raw_input(prompt,echo=True,insert_txt='',use_readline=True):
|
|||
from mmgen.term import kb_hold_protect
|
||||
kb_hold_protect()
|
||||
if echo or not sys.stdin.isatty():
|
||||
reply = input(prompt.encode('utf8'))
|
||||
reply = input(prompt)
|
||||
else:
|
||||
from getpass import getpass
|
||||
reply = getpass(prompt.encode('utf8'))
|
||||
reply = getpass(prompt)
|
||||
kb_hold_protect()
|
||||
|
||||
try:
|
||||
return reply.strip().decode('utf8')
|
||||
return reply.strip()
|
||||
except:
|
||||
die(1,'User input must be UTF-8 encoded.')
|
||||
|
||||
|
|
@ -755,9 +752,9 @@ def prompt_and_get_char(prompt,chars,enter_ok=False,verbose=False):
|
|||
while True:
|
||||
reply = get_char('{}: '.format(prompt)).strip(b'\n\r')
|
||||
|
||||
if reply in chars or (enter_ok and not reply):
|
||||
if reply in chars.encode() or (enter_ok and not reply):
|
||||
msg('')
|
||||
return reply
|
||||
return reply.decode()
|
||||
|
||||
if verbose: msg('\nInvalid reply')
|
||||
else: msg_r('\r')
|
||||
|
|
@ -779,7 +776,7 @@ def do_pager(text):
|
|||
p = Popen([pager],stdin=PIPE,shell=False)
|
||||
except: pass
|
||||
else:
|
||||
p.communicate(text.encode('utf8')+(end_msg,'')[pager=='less'])
|
||||
p.communicate((text+(end_msg,'')[pager=='less']).encode())
|
||||
msg_r('\r')
|
||||
break
|
||||
else: Msg(text+end_msg)
|
||||
|
|
|
|||
|
|
@ -19,7 +19,7 @@ cmd_args = opts.init(opts_data)
|
|||
|
||||
lines = get_lines_from_file(cmd_args[0])
|
||||
start = (1,0)[bool(opt.include_first_line)]
|
||||
a = make_chksum_6(' '.join(lines[start:]).encode('utf8'))
|
||||
a = make_chksum_6(' '.join(lines[start:]).encode())
|
||||
if start == 1:
|
||||
b = lines[0]
|
||||
msg(("Checksum in file ({}) doesn't match computed value!".format(b),'Checksum in file OK')[a==b])
|
||||
|
|
|
|||
|
|
@ -174,8 +174,8 @@ def compile_code(code):
|
|||
cmd = ['solc','--optimize','--bin','--overwrite']
|
||||
if not opt.stdout: cmd += ['--output-dir', opt.outdir or '.']
|
||||
p = Popen(cmd,stdin=PIPE,stdout=PIPE,stderr=PIPE)
|
||||
res = p.communicate(code)
|
||||
o = res[0].replace('\r','').split('\n')
|
||||
res = p.communicate(code.encode())
|
||||
o = res[0].decode().replace('\r','').split('\n')
|
||||
dmsg(res[1])
|
||||
if opt.stdout:
|
||||
return dict((k,o[i+2]) for k in ('SafeMath','Owned','Token') for i in range(len(o)) if k in o[i])
|
||||
|
|
|
|||
2
setup.py
2
setup.py
|
|
@ -18,7 +18,7 @@
|
|||
|
||||
import sys,os,subprocess
|
||||
from shutil import copy2
|
||||
_gvi = subprocess.check_output(['gcc','--version']).splitlines()[0]
|
||||
_gvi = subprocess.check_output(['gcc','--version']).decode().splitlines()[0]
|
||||
have_mingw64 = 'x86_64' in _gvi and 'MinGW' in _gvi
|
||||
have_arm = subprocess.check_output(['uname','-m']).strip() == 'aarch64'
|
||||
|
||||
|
|
|
|||
|
|
@ -81,12 +81,11 @@ if not 1 <= len(cmd_args) <= 2: opts.usage()
|
|||
addr_type = MMGenAddrType(opt.type or g.proto.dfl_mmtype)
|
||||
|
||||
def pyethereum_sec2addr(sec):
|
||||
return sec,eth.privtoaddr(sec).encode('hex')
|
||||
return sec.decode(),eth.privtoaddr(sec).encode('hex')
|
||||
|
||||
def keyconv_sec2addr(sec):
|
||||
p = sp.Popen(['keyconv','-C',g.coin,sec.wif],stderr=sp.PIPE,stdout=sp.PIPE)
|
||||
o = p.stdout.read().splitlines()
|
||||
# print p.stderr.read()
|
||||
o = p.stdout.read().decode().splitlines()
|
||||
return o[1].split()[1],o[0].split()[1]
|
||||
|
||||
def zcash_mini_sec2addr(sec):
|
||||
|
|
@ -97,7 +96,7 @@ def zcash_mini_sec2addr(sec):
|
|||
|
||||
def pycoin_sec2addr(sec):
|
||||
coin = ci.external_tests['testnet']['pycoin'][g.coin] if g.testnet else g.coin
|
||||
key = pcku.parse_key(sec,[network_for_netcode(coin)],secp256k1_generator)[1]
|
||||
key = pcku.parse_key(sec.decode(),[network_for_netcode(coin)],secp256k1_generator)[1]
|
||||
if key is None: die(1,"can't parse {}".format(sec))
|
||||
o = pcku.create_output(sec,key,network_for_netcode(coin))[0]
|
||||
suf = ('_uncompressed','')[addr_type.compressed]
|
||||
|
|
|
|||
|
|
@ -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("'{}'".format(t.decode('utf8').replace('\n',r'\n')))))
|
||||
msg('{}SEND {}{}'.format(ls,es,yellow("'{}'".format(t.replace('\n',r'\n')))))
|
||||
return ret
|
||||
|
||||
def my_expect(p,s,t='',delay=send_delay,regex=False,nonl=False,silent=False):
|
||||
|
|
@ -121,7 +121,7 @@ class MMGenPexpect(object):
|
|||
cmd_str = fs.format(*init_coverage(),c=cmd_str)
|
||||
|
||||
if opt.log:
|
||||
log_fd.write(cmd_str.encode('utf8')+'\n')
|
||||
log_fd.write(cmd_str.encode()+'\n')
|
||||
|
||||
if not no_msg:
|
||||
if opt.verbose or opt.print_cmdline or opt.exact_output:
|
||||
|
|
@ -153,7 +153,7 @@ class MMGenPexpect(object):
|
|||
# PopenSpawn() requires cmd string to be bytes. However, it autoconverts unicode
|
||||
# input to bytes, though this behavior seems to be undocumented. Setting 'encoding'
|
||||
# to 'UTF-8' will cause pexpect to reject non-unicode string input.
|
||||
self.p = PopenSpawn(cmd_str.encode('utf8'))
|
||||
self.p = PopenSpawn(cmd_str,encoding='utf8')
|
||||
else:
|
||||
self.p = pexpect.spawn(cmd,args)
|
||||
if opt.exact_output: self.p.logfile = sys.stdout
|
||||
|
|
@ -245,12 +245,12 @@ class MMGenPexpect(object):
|
|||
if ret == 1:
|
||||
my_send(self.p,'YES\n')
|
||||
# if oo:
|
||||
outfile = self.expect_getend("Overwriting file '").rstrip("'").decode('utf8')
|
||||
outfile = self.expect_getend("Overwriting file '").rstrip("'")
|
||||
return outfile
|
||||
# else:
|
||||
# ret = my_expect(self.p,s1)
|
||||
self.expect(self.NL,nonl=True)
|
||||
outfile = self.p.before.strip().strip("'").decode('utf8')
|
||||
outfile = self.p.before.strip().strip("'")
|
||||
if opt.debug_pexpect: rmsg('Outfile [{}]'.format(outfile))
|
||||
vmsg('{} file: {}'.format(desc,cyan(outfile.replace("'",''))))
|
||||
return outfile
|
||||
|
|
|
|||
|
|
@ -91,7 +91,7 @@ def run_tests():
|
|||
vmsg(green('Executing: {}'.format(cmd)))
|
||||
msg_r('Testing: --coin {:4} {:22}'.format(coin.upper(),type_arg))
|
||||
p = subprocess.Popen(cmd.split(),stdout=subprocess.PIPE,stderr=subprocess.PIPE)
|
||||
o = p.stdout.read()
|
||||
o = p.stdout.read().decode()
|
||||
vmsg(o)
|
||||
o = o.splitlines()
|
||||
d = [e for e in o if len(e) > 4 and e[:9] == 'sc_debug_']
|
||||
|
|
|
|||
|
|
@ -10,7 +10,7 @@ def msg(s): sys.stderr.write(s)
|
|||
def green(s): return '\033[32;1m' + s + '\033[0m'
|
||||
|
||||
def compare_hashes(dlen,data):
|
||||
sha2 = hashlib.sha256(data).hexdigest()
|
||||
sha2 = hashlib.sha256(data).hexdigest().encode()
|
||||
# msg('Dlen {:<5} {}\r'.format(dlen,sha2))
|
||||
my_sha2 = Sha256(data).hexdigest()
|
||||
assert my_sha2 == sha2,'Hashes do not match!'
|
||||
|
|
@ -32,7 +32,7 @@ def test_ref():
|
|||
)
|
||||
for i,data in enumerate(inputs):
|
||||
msg('\rTesting reference input data: {:4}/{} '.format(i+1,len(inputs)))
|
||||
compare_hashes(len(data),data)
|
||||
compare_hashes(len(data),data.encode())
|
||||
msg('OK\n')
|
||||
|
||||
def test_random(rounds):
|
||||
|
|
|
|||
21
test/test.py
21
test/test.py
|
|
@ -1363,7 +1363,7 @@ def create_fake_unspent_entry(coinaddr,al_id=None,idx=None,lbl=None,non_mmgen=Fa
|
|||
if not segwit and k == 'p2sh': k = 'p2pkh'
|
||||
s_beg,s_end = { 'p2pkh': ('76a914','88ac'),
|
||||
'p2sh': ('a914','87'),
|
||||
'bech32': (g.proto.witness_vernum_hex+'14','') }[k]
|
||||
'bech32': (g.proto.witness_vernum_hex.decode()+'14','') }[k]
|
||||
amt1,amt2 = {'btc':(10,40),'bch':(10,40),'ltc':(1000,4000)}[coin_sel]
|
||||
ret = {
|
||||
lbl_id: '{}:{}'.format(g.proto.base_coin.lower(),coinaddr) if non_mmgen \
|
||||
|
|
@ -1373,7 +1373,7 @@ def create_fake_unspent_entry(coinaddr,al_id=None,idx=None,lbl=None,non_mmgen=Fa
|
|||
'amount': g.proto.coin_amt('{}.{}'.format(amt1 + getrandnum(4) % amt2, getrandnum(4) % 100000000)),
|
||||
'address': coinaddr,
|
||||
'spendable': False,
|
||||
'scriptPubKey': '{}{}{}'.format(s_beg,coinaddr.hex,s_end),
|
||||
'scriptPubKey': '{}{}{}'.format(s_beg,coinaddr.hex.decode(),s_end).encode(),
|
||||
'confirmations': getrandnum(3) // 2 # max: 8388608 (7 digits)
|
||||
}
|
||||
return ret
|
||||
|
|
@ -1436,7 +1436,7 @@ def create_fake_unspent_data(adata,tx_data,non_mmgen_input='',non_mmgen_input_co
|
|||
def write_fake_data_to_file(d):
|
||||
unspent_data_file = os.path.join(cfg['tmpdir'],'unspent.json')
|
||||
write_data_to_file(unspent_data_file,d,'Unspent outputs',silent=True,ignore_opt_outdir=True)
|
||||
os.environ['MMGEN_BOGUS_WALLET_DATA'] = unspent_data_file.encode('utf8')
|
||||
os.environ['MMGEN_BOGUS_WALLET_DATA'] = unspent_data_file
|
||||
bwd_msg = 'MMGEN_BOGUS_WALLET_DATA={}'.format(unspent_data_file)
|
||||
if opt.print_cmdline: msg(bwd_msg)
|
||||
if opt.log: log_fd.write(bwd_msg + ' ')
|
||||
|
|
@ -2048,7 +2048,7 @@ class MMGenTestSuite(object):
|
|||
t.written_to_file('Transaction')
|
||||
os.unlink(txfile) # our tx file replaces the original
|
||||
cmd = 'touch ' + os.path.join(cfg['tmpdir'],'txbump')
|
||||
os.system(cmd.encode('utf8'))
|
||||
os.system(cmd.encode())
|
||||
t.ok()
|
||||
|
||||
def txdo(self,name,addrfile,wallet):
|
||||
|
|
@ -2310,7 +2310,7 @@ class MMGenTestSuite(object):
|
|||
self.txsend_ui_common(t,name)
|
||||
|
||||
cmd = 'touch ' + os.path.join(cfg['tmpdir'],'txdo')
|
||||
os.system(cmd.encode('utf8'))
|
||||
os.system(cmd.encode())
|
||||
t.ok()
|
||||
|
||||
def txbump4(self,name,f1,f2,f3,f4,f5,f6,f7,f8,f9): # f7:txfile,f9:'txdo'
|
||||
|
|
@ -3063,7 +3063,7 @@ class MMGenTestSuite(object):
|
|||
['python',os.path.join('cmds','mmgen-tool'),'--testnet=1'] +
|
||||
(['--type=compressed'],[])[i==0] +
|
||||
['-r0','randpair']
|
||||
).split() for i in range(n)]
|
||||
).decode().split() for i in range(n)]
|
||||
restore_debug()
|
||||
return ret
|
||||
|
||||
|
|
@ -3137,7 +3137,7 @@ class MMGenTestSuite(object):
|
|||
|
||||
def regtest_alice_add_label_badaddr(self,name,addr,reply):
|
||||
t = MMGenExpect(name,'mmgen-tool',['--alice','add_label',addr,'(none)'])
|
||||
t.expect(reply.encode('utf8'),regex=True)
|
||||
t.expect(reply,regex=True)
|
||||
t.ok()
|
||||
|
||||
def regtest_alice_add_label_badaddr1(self,name):
|
||||
|
|
@ -3172,7 +3172,7 @@ class MMGenTestSuite(object):
|
|||
|
||||
def regtest_user_chk_label(self,name,user,addr,label,label_pat=None):
|
||||
t = MMGenExpect(name,'mmgen-tool',['--'+user,'listaddresses','all_labels=1'])
|
||||
t.expect(r'{}\s+\S{{30}}\S+\s+{}\s+'.format(addr,(label_pat or label).encode('utf8')),regex=True)
|
||||
t.expect(r'{}\s+\S{{30}}\S+\s+{}\s+'.format(addr,(label_pat or label)),regex=True)
|
||||
t.ok()
|
||||
|
||||
def regtest_alice_chk_label1(self,name):
|
||||
|
|
@ -3479,7 +3479,7 @@ class MMGenTestSuite(object):
|
|||
|
||||
def ethdev_chk_label(self,name,addr='98831F3A:E:3',label_pat=utf8_label_pat):
|
||||
t = MMGenExpect(name,'mmgen-tool', eth_args() + ['listaddresses','all_labels=1'])
|
||||
t.expect(r'{}\s+\S{{30}}\S+\s+{}\s+'.format(addr,(label_pat or label).encode('utf8')),regex=True)
|
||||
t.expect(r'{}\s+\S{{30}}\S+\s+{}\s+'.format(addr,(label_pat or label)),regex=True)
|
||||
t.ok()
|
||||
|
||||
def ethdev_remove_label(self,name,addr='98831F3A:E:3'):
|
||||
|
|
@ -3535,7 +3535,8 @@ class MMGenTestSuite(object):
|
|||
txid = self.txsend_ui_common(t,mmgen_cmd,quiet=True,bogus_send=False,no_ok=True)
|
||||
addr = t.expect_getend('Contract address: ')
|
||||
from mmgen.altcoins.eth.tx import EthereumMMGenTX as etx
|
||||
assert etx.get_exec_status(txid,True) != 0,"Contract '{}:{}' failed to execute. Aborting".format(num,key)
|
||||
assert etx.get_exec_status(txid.encode(),True) != 0,(
|
||||
"Contract '{}:{}' failed to execute. Aborting".format(num,key))
|
||||
if key == 'Token':
|
||||
write_to_tmpfile(cfg,'token_addr{}'.format(num),addr+'\n')
|
||||
silence()
|
||||
|
|
|
|||
|
|
@ -261,9 +261,10 @@ class MMGenToolTestSuite(object):
|
|||
|
||||
p = subprocess.Popen(sys_cmd,stdout=subprocess.PIPE,stderr=subprocess.PIPE)
|
||||
a,b = p.communicate()
|
||||
if not binary: a = a.decode()
|
||||
retcode = p.wait()
|
||||
if retcode != 0:
|
||||
msg('{}\n{}\n{}'.format(red('FAILED'),yellow('Command stderr output:'),b))
|
||||
msg('{}\n{}\n{}'.format(red('FAILED'),yellow('Command stderr output:'),b.decode()))
|
||||
rdie(1,'Called process returned with an error (retcode {})'.format(retcode))
|
||||
return (a,a.rstrip())[bool(strip)]
|
||||
|
||||
|
|
@ -292,7 +293,7 @@ class MMGenToolTestSuite(object):
|
|||
if carg: write_to_tmpfile(cfg,'{}{}.in'.format(name,fn_idx),carg+'\n')
|
||||
ret = self.run_cmd(name,([],[carg])[bool(carg)],kwargs=kwargs,extra_msg=extra_msg,add_opts=add_opts)
|
||||
if carg: vmsg('In: ' + repr(carg))
|
||||
vmsg('Out: ' + (repr(ret),ret.decode('utf8'))[literal])
|
||||
vmsg('Out: ' + (repr(ret),ret)[literal])
|
||||
if ret or ret == '':
|
||||
write_to_tmpfile(cfg,'{}{}.out'.format(name,fn_idx),ret+'\n')
|
||||
if chkdata:
|
||||
|
|
@ -446,7 +447,7 @@ class MMGenToolTestSuite(object):
|
|||
if opt.verbose:
|
||||
sys.stderr.write(green('Executing ') + cyan(cmd) + '\n')
|
||||
p = subprocess.Popen(cmd,stdout=subprocess.PIPE,stderr=subprocess.PIPE,shell=True)
|
||||
res = p.stdout.read().strip()
|
||||
res = p.stdout.read().decode().strip()
|
||||
addr = read_from_tmpfile(cfg,'Wif2addr3.out').strip()
|
||||
cmp_or_die(res,addr)
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue