mmgen-msg: support verification and display of exported JSON data

Usage:

    $ mmgen-msg verify signatures.json

Testing:

    $ test/unit_tests.py -v msg
This commit is contained in:
The MMGen Project 2022-04-28 11:00:52 +00:00
commit 09ab734ab3
Signed by: mmgen
GPG key ID: 3F8B1861E32B7DA2
5 changed files with 78 additions and 11 deletions

View file

@ -33,3 +33,5 @@ class coin_msg(coin_msg):
async def do_verify(self,addr,sig,message):
return await self.rpc.call( 'verifymessage', addr, sig, message )
class exported_sigs(coin_msg.exported_sigs,signed_online): pass

View file

@ -55,8 +55,10 @@ class MsgOps:
class verify(sign):
async def __init__(self,msgfile,addr=None):
m = SignedOnlineMsg( infile=msgfile )
try:
m = SignedOnlineMsg( infile=msgfile )
except:
m = ExportedMsgSigs( infile=msgfile )
qmsg(m.format(addr) + '\n')
@ -81,7 +83,7 @@ opts_data = {
'usage2': [
'[opts] create MESSAGE_TEXT ADDRESS_SPEC [...]',
'[opts] sign MESSAGE_FILE [WALLET_FILE ...]',
'[opts] verify MESSAGE_FILE',
'[opts] verify MESSAGE_FILE or exported JSON dump',
'[opts] export MESSAGE_FILE',
],
'options': """

View file

@ -19,7 +19,7 @@ from .util import msg,vmsg,die,suf,make_chksum_6,fmt_list,remove_dups
from .color import red,orange,grnbg
from .protocol import init_proto
from .fileutil import get_data_from_file,write_data_to_file
from .addr import MMGenID
from .addr import MMGenID,CoinAddr
class MMGenIDRange(str,Hilite,InitErrors,MMGenObject):
"""
@ -87,7 +87,7 @@ class coin_msg:
def get_proto_from_file(self,filename):
data = json.loads(get_data_from_file(filename))
network_id = data['metadata']['network']
network_id = data['metadata']['network'] if 'metadata' in data else data['network'].lower()
coin,network = network_id.split('_')
return init_proto( coin=coin, network=network )
@ -162,7 +162,9 @@ class coin_msg:
yield fs1.format( v[0], v[1](self.data[k]) )
if self.sigs:
yield 'Signature data:'
k = MMGenID(self.proto,req_addr)
k = (
CoinAddr(self.proto,req_addr) if type(self).__name__ == 'exported_sigs' else
MMGenID(self.proto,req_addr) )
if k not in self.sigs:
die(1,f'{k}: address not found in signature data')
for res in gen_entry(self.sigs[k]):
@ -271,7 +273,9 @@ class coin_msg:
def get_sigs(self,addr):
if addr:
req_addr = MMGenID(self.proto,addr)
req_addr = (
CoinAddr(self.proto,addr) if type(self).__name__ == 'exported_sigs' else
MMGenID(self.proto,addr) )
sigs = {k:v for k,v in self.sigs.items() if k == req_addr}
else:
sigs = self.sigs
@ -312,6 +316,22 @@ class coin_msg:
indent = 4
)
class exported_sigs(signed_online):
def __init__(self,infile,*args,**kwargs):
self.data = json.loads(
get_data_from_file(
infile = infile,
desc = self.desc )
)
def gen_sigs():
for e in self.data['signatures']:
yield e
self.sigs = {e['addr']:e for e in gen_sigs()}
def _get_obj(clsname,coin=None,network='mainnet',infile=None,data=None,*args,**kwargs):
assert not args, 'msg:_get_obj(): only keyword args allowed'
@ -348,3 +368,4 @@ CompletedMsg = _get('completed')
UnsignedMsg = _get('unsigned')
SignedMsg = _get('signed')
SignedOnlineMsg = _get('signed_online')
ExportedMsgSigs = _get('exported_sigs')

View file

@ -266,8 +266,10 @@ class TestSuiteRegtest(TestSuiteBase,TestSuiteShared):
('bob_msgverify', 'verifying the message file (all addresses)'),
('bob_msgverify_raw', 'verifying the raw message file (all addresses)'),
('bob_msgverify_single', 'verifying the message file (single address)'),
('bob_msgexport', 'exporting the message file'),
('bob_msgexport_single', 'exporting the message file (single address)'),
('bob_msgexport', 'exporting the message file (all addresses)'),
('bob_msgverify_export', 'verifying the exported JSON data (all addresses)'),
('bob_msgverify_export_single','verifying the exported JSON data (single address)'),
('stop', 'stopping regtest daemon'),
)
@ -1087,13 +1089,13 @@ class TestSuiteRegtest(TestSuiteBase,TestSuiteShared):
fn2 = get_file_with_ext(self.tmpdir,'bip39')
return self.bob_msgsign([fn2,fn1])
def bob_msgverify(self,addr=None,ext='sigmsg.json',cmd='verify'):
def bob_msgverify(self,addr=None,ext='sigmsg.json',cmd='verify',msgfile=None):
return self.spawn(
'mmgen-msg', [
'--bob',
f'--outdir={self.tmpdir}',
cmd,
get_file_with_ext(self.tmpdir,ext),
msgfile or get_file_with_ext(self.tmpdir,ext),
] + ([addr] if addr else []) )
def bob_msgverify_raw(self):
@ -1114,6 +1116,22 @@ class TestSuiteRegtest(TestSuiteBase,TestSuiteShared):
sid = self._user_sid('bob')
return self.bob_msgexport(addr=f'{sid}:{self.dfl_mmtype}:1')
def bob_msgverify_export(self):
return self.bob_msgverify(
msgfile = os.path.join(self.tmpdir,'signatures.json')
)
def bob_msgverify_export_single(self):
sid = self._user_sid('bob')
mmid = f'{sid}:{self.dfl_mmtype}:1'
t = self.spawn('mmgen-tool', [ '--bob', '--color=0', 'listaddress', mmid ], no_msg=True)
addr = t.expect_getend(mmid).split()[0]
t.close()
return self.bob_msgverify(
addr = addr,
msgfile = os.path.join(self.tmpdir,'signatures.json')
)
def stop(self):
if opt.no_daemon_stop:
self.spawn('',msg_only=True)

View file

@ -9,7 +9,8 @@ from test.include.common import silence,end_silence,restart_test_daemons,stop_te
from mmgen.opts import opt
from mmgen.util import msg,bmsg,pumsg
from mmgen.protocol import CoinProtocol
from mmgen.msg import NewMsg,UnsignedMsg,SignedMsg,SignedOnlineMsg
from mmgen.msg import NewMsg,UnsignedMsg,SignedMsg,SignedOnlineMsg,ExportedMsgSigs
from mmgen.addr import MMGenID
def get_obj(coin,network):
@ -65,6 +66,7 @@ async def run_test(network_id):
msg(m.format())
single_addr = 'A091ABAA:111'
single_addr_coin = m.sigs[MMGenID(m.proto,single_addr)]['addr']
pumsg('\nTesting single address display:\n')
msg(m.format(single_addr))
@ -81,6 +83,28 @@ async def run_test(network_id):
pumsg('\nTesting single address JSON dump for export:\n')
msg( m.get_json_for_export(single_addr) )
from mmgen.fileutil import write_data_to_file
exported_sigs = os.path.join(tmpdir,'signatures.json')
write_data_to_file(
outfile = exported_sigs,
data = m.get_json_for_export(),
desc = 'signature data',
ask_overwrite = False )
m = ExportedMsgSigs( infile=exported_sigs )
pumsg('\nTesting verification (exported data):\n')
await m.verify(summary=opt.verbose)
pumsg('\nTesting single address verification (exported data):\n')
await m.verify(single_addr_coin,summary=opt.verbose)
pumsg('\nTesting display (exported data):\n')
msg(m.format())
pumsg('\nTesting single address display (exported data):\n')
msg(m.format(single_addr_coin))
stop_test_daemons(network_id)
msg('\n')