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

new classes: TwJSON.Import, TwJSON.Export

Browse source
This commit is contained in:
The MMGen Project 2022-06-11 16:11:02 +00:00
commit a51352d5b7
Signed by: mmgen
GPG key ID: 3F8B1861E32B7DA2
6 changed files with 265 additions and 139 deletions
Download patch file Download diff file Expand all files Collapse all files

20
mmgen/base_proto/bitcoin/tw/ctl.py
View file

@ -14,7 +14,7 @@ base_proto.bitcoin.twctl: Bitcoin base protocol tracking wallet control class
from ....globalvars import g
from ....tw.ctl import TrackingWallet
from ....util import msg,msg_r,rmsg,vmsg,die,suf,fmt,fmt_list,write_mode,keypress_confirm
from ....util import msg,msg_r,rmsg,vmsg,die,suf,fmt_list,write_mode
class BitcoinTrackingWallet(TrackingWallet):
@ -153,21 +153,3 @@ class BitcoinTrackingWallet(TrackingWallet):
msg('Address has no balance' if len(coin_addrs) == 1 else
'Addresses have no balances' )
return True
async def twimport_check_and_create_wallet(self,info_msg):
if await self.rpc.check_or_create_daemon_wallet(wallet_create=False):
die(3,
f'Existing {self.rpc.daemon.desc} wallet detected!\n' +
'It must be moved, or backed up and securely deleted, before running this command' )
msg('\n'+fmt(info_msg.strip(),indent=' '))
if not keypress_confirm('Continue?'):
msg('Exiting at user request')
return False
if not await self.rpc.check_or_create_daemon_wallet(wallet_create=True):
die(3,'Wallet could not be created')
return True

92
mmgen/base_proto/bitcoin/tw/json.py Executable file
View file

@ -0,0 +1,92 @@
#!/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
"""
base_proto.bitcoin.tw.json: export and import tracking wallet to JSON format
"""
from collections import namedtuple
from ....tw.json import TwJSON
from ....tw.common import TwMMGenID
class BitcoinTwJSON(TwJSON):
class Base(TwJSON.Base):
@property
def mappings_json(self):
return self.json_dump([(e.mmgen_id,e.address) for e in self.entries])
@property
def num_entries(self):
return len(self.entries)
class Import(TwJSON.Import,Base):
info_msg = """
This utility will create a new tracking wallet, import the addresses from
the JSON dump into it and update their balances. The operation may take a
few minutes.
"""
@property
async def tracking_wallet_exists(self):
return await self.tw.rpc.check_or_create_daemon_wallet(wallet_create=False)
async def create_tracking_wallet(self):
return await self.tw.rpc.check_or_create_daemon_wallet(wallet_create=True)
async def get_entries(self):
d = self.data['data']
e_in = [self.entry_tuple_in(*e) for e in d['entries']]
return sorted(
[self.entry_tuple(
TwMMGenID(self.proto,d.mmgen_id),
d.address,
getattr(d,'amount',None),
d.comment)
for d in e_in],
key = lambda x: x.mmgen_id.sort_key )
async def do_import(self,batch):
import_tuple = namedtuple('import_data',['addr','twmmid','comment'])
await self.tw.import_address_common(
[import_tuple(e.address, e.mmgen_id, e.comment) for e in self.entries],
batch = batch )
return [e.address for e in self.entries]
class Export(TwJSON.Export,Base):
@property
async def addrlist(self):
if not hasattr(self,'_addrlist'):
from .addrs import TwAddrList
self._addrlist = await TwAddrList(
proto = self.proto,
usr_addr_list = None,
minconf = 0,
showempty = True,
showbtcaddrs = True,
all_labels = False )
return self._addrlist
async def get_entries(self):
return sorted(
[self.entry_tuple(v['lbl'].mmid, v['addr'], v['amt'], v['lbl'].comment)
for v in (await self.addrlist).values()],
key = lambda x: x.mmgen_id.sort_key )
@property
async def entries_out(self):
return [[getattr(d,k) for k in self.keys] for d in self.entries]
@property
async def total(self):
return (await self.addrlist).total

12
mmgen/tool/rpc.py
View file

@ -194,12 +194,12 @@ class tool_cmd(tool_cmd_base):
async def twexport(self,include_amts=True):
"export the tracking wallet to JSON format"
from ..tw.ctl import TrackingWallet
tw = await TrackingWallet( self.proto )
return await tw.twexport( include_amts=include_amts )
from ..tw.json import TwJSON
await TwJSON.Export( self.proto, include_amts=include_amts )
return True
async def twimport(self,filename:str,ignore_checksum=False,batch=False):
"restore the tracking wallet from a JSON dump created by ‘twexport’"
from ..tw.ctl import TrackingWallet
tw = await TrackingWallet( self.proto, mode='i', rpc_ignore_wallet=True )
return await tw.twimport( filename, ignore_checksum=ignore_checksum, batch=batch )
from ..tw.json import TwJSON
await TwJSON.Import( self.proto, filename, ignore_checksum=ignore_checksum, batch=batch )
return True

114
mmgen/tw/ctl.py
View file

@ -28,24 +28,18 @@ from ..util import (
msg,
msg_r,
qmsg,
ymsg,
dmsg,
suf,
write_mode,
base_proto_subclass,
die,
make_timestamp,
make_chksum_8 )
die )
from ..base_obj import AsyncInit
from ..objmethods import MMGenObject
from ..obj import TwComment,get_obj
from ..addr import CoinAddr,is_mmgen_id,is_coin_addr
from ..rpc import rpc_init,json_encoder
from ..rpc import rpc_init
from .common import TwMMGenID,TwLabel
def json_dump(data):
return json.dumps( data, cls=json_encoder, separators=(',', ':'), sort_keys=True )
class TrackingWallet(MMGenObject,metaclass=AsyncInit):
caps = ('rescan','batch')
@ -53,7 +47,6 @@ class TrackingWallet(MMGenObject,metaclass=AsyncInit):
use_tw_file = False
aggressive_sync = False
importing = False
dump_fn_pfx = 'mmgen-tracking-wallet-dump'
def __new__(cls,proto,*args,**kwargs):
return MMGenObject.__new__(base_proto_subclass(cls,proto,'tw','ctl'))
@ -346,106 +339,3 @@ class TrackingWallet(MMGenObject,metaclass=AsyncInit):
for d in out:
await do_import(*d)
msg('Address import completed OK')
async def twimport(self,filename,ignore_checksum=False,batch=False):
info_msg = """
This utility will create a new tracking wallet, import the addresses from
the JSON dump into it and update their balances. The operation typically
takes just a few minutes.
"""
def check_chksum(d):
chksum = make_chksum_8( json_dump(d['data']).encode() ).lower()
if chksum != d['checksum']:
if ignore_checksum:
ymsg(f'Warning: ignoring incorrect checksum {chksum}')
else:
die(3,'File checksum incorrect! ({} != {})'.format(chksum,d['checksum']))
def check_mappings_chksum(d):
mappings_json = json_dump([e[:2] for e in d['entries']])
mappings_chksum = make_chksum_8(mappings_json.encode()).lower()
if mappings_chksum != d['mappings_checksum']:
die(3,f'ERROR: Mappings checksum incorrect! ({mappings_chksum} != {d["mappings_checksum"]})')
def check_network(d):
coin,network = d['network'].split('_')
if coin != self.proto.coin.lower():
die(2,f'Coin in wallet dump is {coin.upper()}, but configured coin is {self.proto.coin}')
if network != self.proto.network:
die(2,f'Network in wallet dump is {network}, but configured network is {self.proto.network}')
def get_and_verify_data():
from ..fileutil import get_data_from_file
d = json.loads(get_data_from_file(filename,quiet=True))
check_chksum(d)
check_mappings_chksum(d['data'])
check_network(d['data'])
return d
if not await self.twimport_check_and_create_wallet(info_msg):
return True
d = get_and_verify_data()
_d1 = namedtuple('import_data',d['data']['entries_keys'])
entries = [_d1(*e) for e in d['data']['entries']]
_d2 = namedtuple('import_data',['addr','twmmid','comment'])
await self.import_address_common(
[_d2(e.address, e.mmgen_id, e.comment) for e in entries],
batch = batch )
await self.rescan_addresses([e.address for e in entries])
return True
async def twexport(self,include_amts=True):
coin = self.proto.coin_id.lower()
network = self.proto.network
from .addrs import TwAddrList
al = await TwAddrList(
proto = self.proto,
usr_addr_list = None,
minconf = 0,
showempty = True,
showbtcaddrs = True,
all_labels = False )
keys = ['mmgen_id','address'] + (['amount'] if include_amts else []) + ['comment']
entries = sorted(
(
[(v['lbl'].mmid, v['addr'], v['amt'], v['lbl'].comment) for v in al.values()] if include_amts else
[(v['lbl'].mmid, v['addr'], v['lbl'].comment) for v in al.values()]
),
key = lambda x: x[0].sort_key )
mappings_json = json_dump([e[:2] for e in entries])
data = {
'id': 'mmgen_tracking_wallet',
'version': 1,
'network': f'{coin}_{network}',
'blockheight': self.rpc.blockcount,
'time': make_timestamp(),
'mappings_checksum': make_chksum_8(mappings_json.encode()).lower(),
'entries_keys': keys,
'entries': entries,
'num_entries': len(entries),
}
if include_amts:
data['value'] = al.total
chksum = make_chksum_8( json_dump(data).encode() ).lower()
from ..fileutil import write_data_to_file
write_data_to_file(
outfile = f'{self.dump_fn_pfx}-{coin}-{network}.json',
data = json_dump( { 'checksum': chksum, 'data': data } ),
desc = f'tracking wallet JSON data' )
return True

162
mmgen/tw/json.py Executable file
View file

@ -0,0 +1,162 @@
#!/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
"""
tw.json: export and import tracking wallet to JSON format
"""
import json
from collections import namedtuple
from ..util import (
msg,
ymsg,
fmt,
base_proto_subclass,
die,
make_timestamp,
make_chksum_8,
keypress_confirm,
compare_or_die )
from ..base_obj import AsyncInit
from ..objmethods import MMGenObject
from ..rpc import json_encoder
from .ctl import TrackingWallet
class TwJSON:
class Base(MMGenObject):
fn_pfx = 'mmgen-tracking-wallet-dump'
def __new__(cls,proto,*args,**kwargs):
return MMGenObject.__new__(base_proto_subclass(TwJSON,proto,'tw','json',cls.__name__))
def __init__(self,proto):
self.proto = proto
self.coin = proto.coin_id.lower()
self.network = proto.network
self.keys = ['mmgen_id','address','amount','comment']
self.entry_tuple = namedtuple('tw_entry',self.keys)
@property
def dump_fn(self):
return f'{self.fn_pfx}-{self.coin}-{self.network}.json'
def json_dump(self,data):
return json.dumps( data, cls=json_encoder, separators=(',', ':'), sort_keys=True )
def make_chksum(self,data):
return make_chksum_8( self.json_dump(data).encode() ).lower()
@property
def mappings_chksum(self):
return self.make_chksum(self.mappings_json)
@property
def entry_tuple_in(self):
return namedtuple('entry_tuple_in',self.keys)
class Import(Base,metaclass=AsyncInit):
async def __init__(self,proto,filename,ignore_checksum=False,batch=False):
super().__init__(proto)
self.tw = await TrackingWallet( proto, mode='i', rpc_ignore_wallet=True )
def check_network(data):
coin,network = data['network'].split('_')
if coin != self.coin:
die(2,f'Coin in wallet dump is {coin.upper()}, but configured coin is {self.coin.upper()}')
if network != self.network:
die(2,f'Network in wallet dump is {network}, but configured network is {self.network}')
def check_chksum(d):
chksum = self.make_chksum(d['data'])
if chksum != d['checksum']:
if ignore_checksum:
ymsg(f'Warning: ignoring incorrect checksum {chksum}')
else:
die(3,'File checksum incorrect! ({} != {})'.format(chksum,d['checksum']))
def verify_data(d):
check_network(d['data'])
check_chksum(d)
compare_or_die(
self.mappings_chksum, 'computed mappings checksum',
d['data']['mappings_checksum'], 'saved checksum' )
if not await self.check_and_create_wallet():
return True
from ..fileutil import get_data_from_file
self.data = json.loads(get_data_from_file(filename,quiet=True))
self.keys = self.data['data']['entries_keys']
self.entries = await self.get_entries()
verify_data(self.data)
addrs = await self.do_import(batch)
await self.tw.rescan_addresses(addrs)
async def check_and_create_wallet(self):
if await self.tracking_wallet_exists:
die(3,
f'Existing {self.tw.rpc.daemon.desc} wallet detected!\n' +
'It must be moved, or backed up and securely deleted, before running this command' )
msg('\n'+fmt(self.info_msg.strip(),indent=' '))
if not keypress_confirm('Continue?'):
msg('Exiting at user request')
return False
if not await self.create_tracking_wallet():
die(3,'Wallet could not be created')
return True
class Export(Base,metaclass=AsyncInit):
async def __init__(self,proto,include_amts=True):
super().__init__(proto)
if not include_amts:
self.keys.remove('amount')
self.tw = await TrackingWallet( proto )
self.entries = await self.get_entries()
data = {
'id': 'mmgen_tracking_wallet',
'version': 1,
'network': f'{self.coin}_{self.network}',
'blockheight': self.tw.rpc.blockcount,
'time': make_timestamp(),
'mappings_checksum': self.mappings_chksum,
'entries_keys': self.keys,
'entries': await self.entries_out,
'num_entries': self.num_entries,
}
if include_amts:
data['value'] = await self.total
from ..fileutil import write_data_to_file
write_data_to_file(
outfile = self.dump_fn,
data = self.json_dump({
'checksum': self.make_chksum(data),
'data': data }),
desc = f'tracking wallet JSON data' )

4
test/test_py_d/ts_regtest.py
View file

@ -987,8 +987,8 @@ class TestSuiteRegtest(TestSuiteBase,TestSuiteShared):
return self.bob_twexport(add_args=['include_amts=0'])
def carol_twimport(self,add_args=[]):
from mmgen.tw.ctl import TrackingWallet as twcls
fn = joinpath(self.tmpdir,f'{twcls.dump_fn_pfx}-{self.proto.coin.lower()}-regtest.json')
from mmgen.tw.json import TwJSON
fn = joinpath( self.tmpdir, TwJSON.Base(self.proto).dump_fn )
t = self.spawn('mmgen-tool',['--carol','twimport',fn] + add_args)
t.expect('(y/N): ','y')
if 'batch=true' in add_args: