From 5961d1c36d0f53293c047fc58a4626d2b6d5b7e4 Mon Sep 17 00:00:00 2001 From: The MMGen Project Date: Sat, 15 Jan 2022 14:00:08 +0000 Subject: [PATCH] addr.py: move AddrData and TwAddrData classes to addrdata.py --- mmgen/addr.py | 94 ------------------------------ mmgen/addrdata.py | 119 ++++++++++++++++++++++++++++++++++++++ mmgen/altcoins/eth/tw.py | 2 +- mmgen/tw.py | 4 +- mmgen/tx.py | 5 +- test/test_py_d/ts_main.py | 3 +- 6 files changed, 127 insertions(+), 100 deletions(-) create mode 100755 mmgen/addrdata.py diff --git a/mmgen/addr.py b/mmgen/addr.py index 1d522473..f34c9cd3 100755 --- a/mmgen/addr.py +++ b/mmgen/addr.py @@ -828,97 +828,3 @@ Record this checksum: it will be used to verify the password file in the future from .crypto import scramble_seed dmsg_sc('str',scramble_key) return scramble_seed(seed,scramble_key.encode()) - -class AddrData(MMGenObject): - msgs = { - 'too_many_acct_addresses': f""" -ERROR: More than one address found for account: '{{}}'. -Your 'wallet.dat' file appears to have been altered by a non-{pnm} program. -Please restore your tracking wallet from a backup or create a new one and -re-import your addresses. -""".strip() - } - - def __new__(cls,proto,*args,**kwargs): - return MMGenObject.__new__(altcoin_subclass(cls,proto,'tw')) - - def __init__(self,proto,*args,**kwargs): - self.al_ids = {} - self.proto = proto - self.rpc = None - - def seed_ids(self): - return list(self.al_ids.keys()) - - def addrlist(self,al_id): - # TODO: Validate al_id - if al_id in self.al_ids: - return self.al_ids[al_id] - - def mmaddr2coinaddr(self,mmaddr): - al_id,idx = MMGenID(self.proto,mmaddr).rsplit(':',1) - coinaddr = '' - if al_id in self.al_ids: - coinaddr = self.addrlist(al_id).coinaddr(int(idx)) - return coinaddr or None - - def coinaddr2mmaddr(self,coinaddr): - d = self.make_reverse_dict([coinaddr]) - return (list(d.values())[0][0]) if d else None - - def add(self,addrlist): - if type(addrlist) == AddrList: - self.al_ids[addrlist.al_id] = addrlist - return True - else: - raise TypeError(f'Error: object {addrlist!r} is not of type AddrList') - - def make_reverse_dict(self,coinaddrs): - d = MMGenDict() - for al_id in self.al_ids: - d.update(self.al_ids[al_id].make_reverse_dict_addrlist(coinaddrs)) - return d - -class TwAddrData(AddrData,metaclass=AsyncInit): - - def __new__(cls,proto,*args,**kwargs): - return MMGenObject.__new__(altcoin_subclass(cls,proto,'tw')) - - async def __init__(self,proto,wallet=None): - self.proto = proto - from .rpc import rpc_init - self.rpc = await rpc_init(proto) - self.al_ids = {} - await self.add_tw_data(wallet) - - async def get_tw_data(self,wallet=None): - vmsg('Getting address data from tracking wallet') - c = self.rpc - if 'label_api' in c.caps: - accts = await c.call('listlabels') - ll = await c.batch_call('getaddressesbylabel',[(k,) for k in accts]) - alists = [list(a.keys()) for a in ll] - else: - accts = await c.call('listaccounts',0,True) - alists = await c.batch_call('getaddressesbyaccount',[(k,) for k in accts]) - return list(zip(accts,alists)) - - async def add_tw_data(self,wallet): - - twd = await self.get_tw_data(wallet) - out,i = {},0 - for acct,addr_array in twd: - l = get_obj(TwLabel,proto=self.proto,text=acct,silent=True) - if l and l.mmid.type == 'mmgen': - obj = l.mmid.obj - if len(addr_array) != 1: - die(2,self.msgs['too_many_acct_addresses'].format(acct)) - al_id = AddrListID(SeedID(sid=obj.sid),self.proto.addr_type(obj.mmtype)) - if al_id not in out: - out[al_id] = [] - out[al_id].append(AddrListEntry(self.proto,idx=obj.idx,addr=addr_array[0],label=l.comment)) - i += 1 - - vmsg(f'{i} {pnm} addresses found, {len(twd)} accounts total') - for al_id in out: - self.add(AddrList(self.proto,al_id=al_id,adata=AddrListData(sorted(out[al_id],key=lambda a: a.idx)))) diff --git a/mmgen/addrdata.py b/mmgen/addrdata.py new file mode 100755 index 00000000..346dff70 --- /dev/null +++ b/mmgen/addrdata.py @@ -0,0 +1,119 @@ +#!/usr/bin/env python3 +# +# mmgen = Multi-Mode GENerator, command-line Bitcoin cold storage solution +# Copyright (C)2013-2022 The MMGen Project +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . + +""" +addrdata.py: MMGen AddrData and related classes +""" + +from .util import vmsg,altcoin_subclass +from .base_obj import AsyncInit +from .obj import MMGenObject,MMGenDict,get_obj,AddrListID,AddrListData +from .addr import MMGenID,AddrListEntry,AddrList + +class AddrData(MMGenObject): + msgs = { + 'too_many_acct_addresses': """ +ERROR: More than one address found for account: '{}'. +Your 'wallet.dat' file appears to have been altered by a non-{} program. +Please restore your tracking wallet from a backup or create a new one and +re-import your addresses. +""".strip() + } + + def __new__(cls,proto,*args,**kwargs): + return MMGenObject.__new__(altcoin_subclass(cls,proto,'tw')) + + def __init__(self,proto,*args,**kwargs): + self.al_ids = {} + self.proto = proto + self.rpc = None + + def seed_ids(self): + return list(self.al_ids.keys()) + + def addrlist(self,al_id): + # TODO: Validate al_id + if al_id in self.al_ids: + return self.al_ids[al_id] + + def mmaddr2coinaddr(self,mmaddr): + al_id,idx = MMGenID(self.proto,mmaddr).rsplit(':',1) + coinaddr = '' + if al_id in self.al_ids: + coinaddr = self.addrlist(al_id).coinaddr(int(idx)) + return coinaddr or None + + def coinaddr2mmaddr(self,coinaddr): + d = self.make_reverse_dict([coinaddr]) + return (list(d.values())[0][0]) if d else None + + def add(self,addrlist): + if type(addrlist) == AddrList: + self.al_ids[addrlist.al_id] = addrlist + return True + else: + raise TypeError(f'Error: object {addrlist!r} is not of type AddrList') + + def make_reverse_dict(self,coinaddrs): + d = MMGenDict() + for al_id in self.al_ids: + d.update(self.al_ids[al_id].make_reverse_dict_addrlist(coinaddrs)) + return d + +class TwAddrData(AddrData,metaclass=AsyncInit): + + def __new__(cls,proto,*args,**kwargs): + return MMGenObject.__new__(altcoin_subclass(cls,proto,'tw')) + + async def __init__(self,proto,wallet=None): + from .rpc import rpc_init + from .obj import TwLabel + from .globalvars import g + from .seed import SeedID + self.proto = proto + self.rpc = await rpc_init(proto) + self.al_ids = {} + twd = await self.get_tw_data(wallet) + out,i = {},0 + for acct,addr_array in twd: + l = get_obj(TwLabel,proto=self.proto,text=acct,silent=True) + if l and l.mmid.type == 'mmgen': + obj = l.mmid.obj + if len(addr_array) != 1: + die(2,self.msgs['too_many_acct_addresses'].format(acct,g.prog_name)) + al_id = AddrListID(SeedID(sid=obj.sid),self.proto.addr_type(obj.mmtype)) + if al_id not in out: + out[al_id] = [] + out[al_id].append(AddrListEntry(self.proto,idx=obj.idx,addr=addr_array[0],label=l.comment)) + i += 1 + + vmsg(f'{i} {g.prog_name} addresses found, {len(twd)} accounts total') + for al_id in out: + self.add(AddrList(self.proto,al_id=al_id,adata=AddrListData(sorted(out[al_id],key=lambda a: a.idx)))) + + async def get_tw_data(self,wallet=None): + vmsg('Getting address data from tracking wallet') + c = self.rpc + if 'label_api' in c.caps: + accts = await c.call('listlabels') + ll = await c.batch_call('getaddressesbylabel',[(k,) for k in accts]) + alists = [list(a.keys()) for a in ll] + else: + accts = await c.call('listaccounts',0,True) + alists = await c.batch_call('getaddressesbyaccount',[(k,) for k in accts]) + return list(zip(accts,alists)) diff --git a/mmgen/altcoins/eth/tw.py b/mmgen/altcoins/eth/tw.py index 1b2aa0ae..3377e670 100755 --- a/mmgen/altcoins/eth/tw.py +++ b/mmgen/altcoins/eth/tw.py @@ -23,7 +23,7 @@ altcoins.eth.tw: Ethereum tracking wallet and related classes for the MMGen suit from mmgen.common import * from mmgen.obj import TwLabel,is_coin_addr,is_mmgen_id,ListItemAttr,ImmutableAttr from mmgen.tw import TrackingWallet,TwAddrList,TwUnspentOutputs,TwGetBalance -from mmgen.addr import AddrData,TwAddrData +from mmgen.addrdata import AddrData,TwAddrData from .contract import Token,TokenResolve from .obj import ETHAmt diff --git a/mmgen/tw.py b/mmgen/tw.py index 33e24aef..9f15348b 100755 --- a/mmgen/tw.py +++ b/mmgen/tw.py @@ -938,7 +938,7 @@ class TrackingWallet(MMGenObject,metaclass=AsyncInit): mmaddr = TwMMGenID(self.proto,arg1) if mmaddr and not coinaddr: - from .addr import TwAddrData + from .addrdata import TwAddrData coinaddr = (await TwAddrData(self.proto)).mmaddr2coinaddr(mmaddr) try: @@ -953,7 +953,7 @@ class TrackingWallet(MMGenObject,metaclass=AsyncInit): # Allow for the possibility that BTC addr of MMGen addr was entered. # Do reverse lookup, so that MMGen addr will not be marked as non-MMGen. if not mmaddr: - from .addr import TwAddrData + from .addrdata import TwAddrData mmaddr = (await TwAddrData(proto=self.proto)).coinaddr2mmaddr(coinaddr) if not mmaddr: diff --git a/mmgen/tx.py b/mmgen/tx.py index ef100e39..323467a2 100755 --- a/mmgen/tx.py +++ b/mmgen/tx.py @@ -723,7 +723,8 @@ class MMGenTX: die(2,'At least one output must be specified on the command line') async def get_outputs_from_cmdline(self,cmd_args): - from .addr import AddrList,AddrData,TwAddrData + from .addr import AddrList + from .addrdata import AddrData,TwAddrData from .addrfile import AddrFile addrfiles = remove_dups( tuple(a for a in cmd_args if get_extension(a) == AddrFile.ext), @@ -1612,7 +1613,7 @@ class MMGenTX: # # async def get_outputs_from_cmdline(self,mmid): # TODO: check that addr is empty # -# from .addr import TwAddrData +# from .addrdata import TwAddrData # ad_w = await TwAddrData() # # if is_mmgen_id(self.proto,mmid): diff --git a/test/test_py_d/ts_main.py b/test/test_py_d/ts_main.py index 51b3e748..31c845e7 100755 --- a/test/test_py_d/ts_main.py +++ b/test/test_py_d/ts_main.py @@ -397,8 +397,9 @@ class TestSuiteMain(TestSuiteBase,TestSuiteShared): return out def _create_tx_data(self,sources,addrs_per_wallet=addrs_per_wallet): - from mmgen.addr import AddrData,AddrList + from mmgen.addr import AddrList from mmgen.obj import AddrIdxList + from mmgen.addrdata import AddrData tx_data,ad = {},AddrData(self.proto) for s in sources: afile = get_file_with_ext(self.cfgs[s]['tmpdir'],'addrs')