new method: proto.parse_addr(); new property: CoinAddr.parsed
This commit is contained in:
parent
a48b43c838
commit
dc3a6b8d49
5 changed files with 110 additions and 2 deletions
|
|
@ -147,13 +147,19 @@ class CoinAddr(str,Hilite,InitErrors,MMGenObject):
|
|||
ap = proto.decode_addr(addr)
|
||||
assert ap, f'coin address {addr!r} could not be parsed'
|
||||
me.addr_fmt = ap.fmt
|
||||
me.hex = ap.bytes.hex()
|
||||
me.bytes = ap.bytes
|
||||
me.ver_bytes = ap.ver_bytes
|
||||
me.proto = proto
|
||||
return me
|
||||
except Exception as e:
|
||||
return cls.init_fail(e,addr,objname=f'{proto.cls_name} address')
|
||||
|
||||
@property
|
||||
def parsed(self):
|
||||
if not hasattr(self,'_parsed'):
|
||||
self._parsed = self.proto.parse_addr(self.ver_bytes,self.bytes,self.addr_fmt)
|
||||
return self._parsed
|
||||
|
||||
@classmethod
|
||||
def fmtc(cls,addr,**kwargs):
|
||||
w = kwargs['width'] or cls.width
|
||||
|
|
|
|||
|
|
@ -12,8 +12,12 @@
|
|||
Monero protocol
|
||||
"""
|
||||
|
||||
from collections import namedtuple
|
||||
|
||||
from ..protocol import CoinProtocol,_nw
|
||||
|
||||
parsed_addr = namedtuple('parsed_addr',['ver_bytes','data'])
|
||||
|
||||
# https://github.com/monero-project/monero/blob/master/src/cryptonote_config.h
|
||||
class mainnet(CoinProtocol.DummyWIF,CoinProtocol.Base):
|
||||
|
||||
|
|
@ -59,6 +63,13 @@ class mainnet(CoinProtocol.DummyWIF,CoinProtocol.Base):
|
|||
|
||||
return self.decode_addr_bytes(ret[:-4])
|
||||
|
||||
def parse_addr(self,ver_bytes,addr_bytes,fmt):
|
||||
addr_len = self.get_addr_len('monero')
|
||||
return parsed_addr(
|
||||
ver_bytes = ver_bytes,
|
||||
data = addr_bytes[:addr_len],
|
||||
)
|
||||
|
||||
def pubhash2addr(self,*args,**kwargs):
|
||||
raise NotImplementedError('Monero addresses do not support pubhash2addr()')
|
||||
|
||||
|
|
|
|||
|
|
@ -27,6 +27,7 @@ from .globalvars import g
|
|||
|
||||
parsed_wif = namedtuple('parsed_wif',['sec','pubkey_type','compressed'])
|
||||
decoded_addr = namedtuple('decoded_addr',['bytes','ver_bytes','fmt'])
|
||||
parsed_addr = namedtuple('parsed_addr',['ver_bytes','data'])
|
||||
|
||||
_finfo = namedtuple('fork_info',['height','hash','name','replayable'])
|
||||
_nw = namedtuple('coin_networks',['mainnet','testnet','regtest'])
|
||||
|
|
@ -163,6 +164,12 @@ class CoinProtocol(MMGenObject):
|
|||
privkey_len = 32
|
||||
pubkey_types = ('std',)
|
||||
|
||||
def parse_addr(self,ver_bytes,addr_bytes,fmt):
|
||||
return parsed_addr(
|
||||
ver_bytes = ver_bytes,
|
||||
data = addr_bytes,
|
||||
)
|
||||
|
||||
def preprocess_key(self,sec,pubkey_type):
|
||||
# Key must be non-zero and less than group order of secp256k1 curve
|
||||
if 0 < int.from_bytes(sec,'big') < self.secp256k1_ge:
|
||||
|
|
|
|||
|
|
@ -359,7 +359,7 @@ class TestSuiteMain(TestSuiteBase,TestSuiteShared):
|
|||
getrandnum(4) % 100000000 )),
|
||||
'address': coinaddr,
|
||||
'spendable': False,
|
||||
'scriptPubKey': f'{s_beg}{coinaddr.hex}{s_end}',
|
||||
'scriptPubKey': f'{s_beg}{coinaddr.bytes.hex()}{s_end}',
|
||||
'confirmations': getrandnum(3) // 20 # max: 838860 (6 digits)
|
||||
}
|
||||
return ret
|
||||
|
|
|
|||
84
test/unit_tests_d/ut_addrparse.py
Executable file
84
test/unit_tests_d/ut_addrparse.py
Executable file
|
|
@ -0,0 +1,84 @@
|
|||
#!/usr/bin/env python3
|
||||
"""
|
||||
test/unit_tests_d/ut_addrparse: address parsing tests for the MMGen suite
|
||||
"""
|
||||
|
||||
from mmgen.common import *
|
||||
|
||||
vectors = {
|
||||
'btc_mainnet': [
|
||||
{'std': '1C5VPtgq9xQ6AcTgMAR3J6GDrs72HC4pS1'},
|
||||
{'std': '3AhjTiWHhVJAi1s5CfKMcLzYps12x3gZhg'},
|
||||
{'std': 'bc1q6pqnfwwakuuejpm9w52ds342f9d5u36v0qnz7c'}
|
||||
],
|
||||
'ltc_mainnet': [
|
||||
{'std': 'LUbHQNYoy23RByq4dKQotLA4ugk9FhpAMT'},
|
||||
{'std': 'MCoZrHYPqYKqvpiwyzzqf3EPxF5no6puEf'},
|
||||
{'std': 'ltc1qvmqas4maw7lg9clqu6kqu9zq9cluvllnst5pxs'}
|
||||
],
|
||||
'xmr_mainnet': [
|
||||
{ # ut_xmrseed.vectors[0]:
|
||||
'std': '42ey1afDFnn4886T7196doS9GPMzexD9gXpsZJDwVjeRVdFCSoHnv7KPbBeGpzJBzHRCAs9UxqeoyFQMYbqSWYTfJJQAWDm',
|
||||
},{
|
||||
'std': '46r4nYSevkfBUMhuykdK3gQ98XDqDTYW1hNLaXNvjpsJaSbNtdXh1sKMsdVgqkaihChAzEy29zEDPMR3NHQvGoZCLGwTerK',
|
||||
}
|
||||
],
|
||||
'zec_mainnet': [
|
||||
{'std': 't1KQYLBvjpmcQuATommo6gx2QTQDLPikB8Q'},
|
||||
{'std': 'zceQDpyNwek7dKqF5ZuFGj7YrNVxh7X1aPkrVxDLVxWSiZAFDEuy5C7XNV8VhyZ3ghTPQ61xjCGiyLT3wqpiN1Yi6mdmaCq'},
|
||||
],
|
||||
'eth_mainnet': [
|
||||
{'std': '7e5f4552091a69125d5dfcb7b8c2659029395bdf'},
|
||||
],
|
||||
}
|
||||
|
||||
def test_network(proto,addrs):
|
||||
|
||||
def check_equal(a,b):
|
||||
assert a == b, f'{a.hex()} != {b.hex()}'
|
||||
|
||||
def check_bytes(addr):
|
||||
if addr.parsed.ver_bytes is not None:
|
||||
check_equal(
|
||||
addr.parsed.ver_bytes,
|
||||
proto.addr_fmt_to_ver_bytes.get(addr.addr_fmt) )
|
||||
check_equal(
|
||||
addr.parsed.data,
|
||||
addr.bytes )
|
||||
|
||||
def fmt_addr_data(addr):
|
||||
return pp_fmt({k:(v.hex() if isinstance(v,bytes) else v) for k,v in addr.parsed._asdict().items()})
|
||||
|
||||
def print_info(addr):
|
||||
vmsg('\n{}\n{}\n{}'.format(
|
||||
yellow(addr.addr_fmt),
|
||||
cyan(addr),
|
||||
fmt_addr_data(addr)))
|
||||
|
||||
msg_r(f'Testing {proto.coin} address parsing...')
|
||||
vmsg('')
|
||||
|
||||
from mmgen.addr import CoinAddr
|
||||
|
||||
for addr in addrs:
|
||||
a1 = CoinAddr(proto,addr['std'])
|
||||
print_info(a1)
|
||||
check_bytes(a1)
|
||||
|
||||
msg('OK')
|
||||
vmsg('')
|
||||
|
||||
|
||||
class unit_test(object):
|
||||
|
||||
def run_test(self,name,ut):
|
||||
|
||||
from mmgen.protocol import init_proto
|
||||
|
||||
for net_id,addrs in vectors.items():
|
||||
coin,network = net_id.split('_')
|
||||
test_network(
|
||||
init_proto(coin,network=network),
|
||||
addrs )
|
||||
|
||||
return True
|
||||
Loading…
Add table
Add a link
Reference in a new issue