diff --git a/bech32m.py b/bech32m.py new file mode 100644 index 0000000..f79aacf --- /dev/null +++ b/bech32m.py @@ -0,0 +1,137 @@ +# Copyright (c) 2017, 2020 Pieter Wuille +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +# THE SOFTWARE. + +"""Reference implementation for Bech32/Bech32m and segwit addresses.""" + +from enum import Enum + +class Encoding(Enum): + """Enumeration type to list the various supported encodings.""" + BECH32 = 1 + BECH32M = 2 + +CHARSET = "qpzry9x8gf2tvdw0s3jn54khce6mua7l" +BECH32M_CONST = 0x2bc830a3 + +def bech32_polymod(values): + """Internal function that computes the Bech32 checksum.""" + generator = [0x3b6a57b2, 0x26508e6d, 0x1ea119fa, 0x3d4233dd, 0x2a1462b3] + chk = 1 + for value in values: + top = chk >> 25 + chk = (chk & 0x1ffffff) << 5 ^ value + for i in range(5): + chk ^= generator[i] if ((top >> i) & 1) else 0 + return chk + + +def bech32_hrp_expand(hrp): + """Expand the HRP into values for checksum computation.""" + return [ord(x) >> 5 for x in hrp] + [0] + [ord(x) & 31 for x in hrp] + + +def bech32_verify_checksum(hrp, data): + """Verify a checksum given HRP and converted data characters.""" + const = bech32_polymod(bech32_hrp_expand(hrp) + data) + if const == 1: + return Encoding.BECH32 + if const == BECH32M_CONST: + return Encoding.BECH32M + return None + +def bech32_create_checksum(hrp, data, spec): + """Compute the checksum values given HRP and data.""" + values = bech32_hrp_expand(hrp) + data + const = BECH32M_CONST if spec == Encoding.BECH32M else 1 + polymod = bech32_polymod(values + [0, 0, 0, 0, 0, 0]) ^ const + return [(polymod >> 5 * (5 - i)) & 31 for i in range(6)] + + +def bech32_encode(hrp, data, spec): + """Compute a Bech32 string given HRP and data values.""" + combined = data + bech32_create_checksum(hrp, data, spec) + return hrp + '1' + ''.join([CHARSET[d] for d in combined]) + +def bech32_decode(bech): + """Validate a Bech32/Bech32m string, and determine HRP and data.""" + if ((any(ord(x) < 33 or ord(x) > 126 for x in bech)) or + (bech.lower() != bech and bech.upper() != bech)): + return (None, None, None) + bech = bech.lower() + pos = bech.rfind('1') + if pos < 1 or pos + 7 > len(bech): + return (None, None, None) + if not all(x in CHARSET for x in bech[pos+1:]): + return (None, None, None) + hrp = bech[:pos] + data = [CHARSET.find(x) for x in bech[pos+1:]] + spec = bech32_verify_checksum(hrp, data) + if spec is None: + return (None, None, None) + return (hrp, data[:-6], spec) + +def convertbits(data, frombits, tobits, pad=True): + """General power-of-2 base conversion.""" + acc = 0 + bits = 0 + ret = [] + maxv = (1 << tobits) - 1 + max_acc = (1 << (frombits + tobits - 1)) - 1 + for value in data: + if value < 0 or (value >> frombits): + return None + acc = ((acc << frombits) | value) & max_acc + bits += frombits + while bits >= tobits: + bits -= tobits + ret.append((acc >> bits) & maxv) + if pad: + if bits: + ret.append((acc << (tobits - bits)) & maxv) + elif bits >= frombits or ((acc << (tobits - bits)) & maxv): + return None + return ret + + +def decode(hrp, addr): + """Decode a segwit address.""" + hrpgot, data, spec = bech32_decode(addr) + if hrpgot != hrp: + return (None, None) + decoded = convertbits(data[1:], 5, 8, False) + if decoded is None or len(decoded) < 2 or len(decoded) > 40: + return (None, None) + if data[0] > 16: + return (None, None) + if data[0] == 0 and len(decoded) != 20 and len(decoded) != 32: + return (None, None) + if data[0] == 0 and spec != Encoding.BECH32 or data[0] != 0 and spec != Encoding.BECH32M: + return (None, None) + return (data[0], decoded) + + +def encode(hrp, witver, witprog): + """Encode a segwit address.""" + spec = Encoding.BECH32 if witver == 0 else Encoding.BECH32M + ret = bech32_encode(hrp, [witver] + convertbits(witprog, 8, 5), spec) + if decode(hrp, ret) == (None, None): + return None + return ret + diff --git a/unified_addrs.py b/unified_addrs.py new file mode 100644 index 0000000..e967861 --- /dev/null +++ b/unified_addrs.py @@ -0,0 +1,178 @@ +#!/usr/bin/env python3 +import sys; assert sys.version_info[0] >= 3, "Python 3 required." + +import math +import struct + +from pyblake2 import blake2b +from bech32m import bech32_encode, bech32_decode, convertbits, Encoding + +from tv_output import render_args, render_tv, Some +from tv_rand import Rand +from f4jumble import f4jumble, f4jumble_inv +import sapling_key_components +import orchard_key_components + +def tlv(typecode, value): + return b"".join([bytes([typecode, len(value)]), value]) + +def encode_unified(receivers): + orchard_receiver = b"" + if receivers[0]: + orchard_receiver = tlv(0x03, receivers[0]) + + sapling_receiver = b"" + if receivers[1]: + sapling_receiver = tlv(0x02, receivers[1]) + + t_receiver = b"" + if receivers[2][1]: + if receivers[2][0]: + typecode = 0x00 + else: + typecode = 0x01 + t_receiver = tlv(typecode, receivers[2][1]) + + r_bytes = b"".join([orchard_receiver, sapling_receiver, t_receiver, bytes(16)]) + converted = convertbits(f4jumble(r_bytes), 8, 5) + return bech32_encode("u", converted, Encoding.BECH32M) + +def decode_unified(addr_str): + (hrp, data, encoding) = bech32_decode(addr_str) + assert hrp == "u" and encoding == Encoding.BECH32M + + decoded = f4jumble_inv(bytes(convertbits(data, 5, 8, False))) + suffix = decoded[-16:] + # check trailing zero bytes + assert suffix == bytes(16) + decoded = decoded[:-16] + + s = 0 + acc = [] + result = {} + for b in decoded: + if s == 0: + receiver_type = b + s = 1 + elif s == 1: + receiver_len = b + expected_len = {0: 20, 1: 20, 2: 43, 3: 43}.get(receiver_type) + if expected_len is not None: + assert receiver_len == expected_len, "incorrect receiver length" + s = 2 + elif s == 2: + if len(acc) < receiver_len: + acc.append(b) + + if len(acc) == receiver_len: + if receiver_type == 0 or receiver_type == 1: + assert not ('transparent' in result), "duplicate transparent receiver detected" + assert len(acc) == 20 + result['transparent'] = bytes(acc) + acc = [] + s = 0 + + elif receiver_type == 2: + assert not ('sapling' in result), "duplicate sapling receiver detected" + assert len(acc) == 43 + result['sapling'] = bytes(acc) + acc = [] + s = 0 + + elif receiver_type == 3: + assert not ('orchard' in result), "duplicate orchard receiver detected" + assert len(acc) == 43 + result['orchard'] = bytes(acc) + acc = [] + s = 0 + return result + + +def main(): + args = render_args() + + from random import Random + rng = Random(0xabad533d) + def randbytes(l): + ret = [] + while len(ret) < l: + ret.append(rng.randrange(0, 256)) + return bytes(ret) + rand = Rand(randbytes) + + test_vectors = [] + for _ in range(0, 10): + has_t_addr = rand.bool() + if has_t_addr: + t_addr = b"".join([rand.b(20)]) + else: + t_addr = None + + has_s_addr = rand.bool() + if has_s_addr: + sapling_sk = sapling_key_components.SpendingKey(rand.b(32)) + sapling_default_d = sapling_sk.default_d() + sapling_default_pk_d = sapling_sk.default_pkd() + sapling_raw_addr = b"".join([sapling_default_d[:11], bytes(sapling_default_pk_d)[:32]]) + else: + sapling_raw_addr = None + + has_o_addr = (not has_s_addr) or rand.bool() + if has_o_addr: + orchard_sk = orchard_key_components.SpendingKey(rand.b(32)) + orchard_fvk = orchard_key_components.FullViewingKey(orchard_sk) + orchard_default_d = orchard_fvk.default_d() + orchard_default_pk_d = orchard_fvk.default_pkd() + orchard_raw_addr = b"".join([orchard_default_d[:11], bytes(orchard_default_pk_d)[:32]]) + else: + orchard_raw_addr = None + + is_p2pkh = rand.bool() + receivers = [ + orchard_raw_addr, + sapling_raw_addr, + (is_p2pkh, t_addr) + ] + ua = encode_unified(receivers) + + decoded = decode_unified(ua) + assert decoded.get('orchard') == orchard_raw_addr + assert decoded.get('sapling') == sapling_raw_addr + assert decoded.get('transparent') == t_addr + + test_vectors.append({ + 'p2pkh_bytes': t_addr if is_p2pkh else None, + 'p2sh_bytes': None if is_p2pkh else t_addr, + 'sapling_raw_addr': sapling_raw_addr, + 'orchard_raw_addr': orchard_raw_addr, + 'unified_addr': ua.encode() + }) + + render_tv( + args, + 'unified_address', + ( + ('p2pkh_bytes', { + 'rust_type': 'Option<[u8; 20]>', + 'rust_fmt': lambda x: None if x is None else Some(x), + }), + ('p2sh_bytes', { + 'rust_type': 'Option<[u8; 20]>', + 'rust_fmt': lambda x: None if x is None else Some(x), + }), + ('sapling_raw_addr', { + 'rust_type': 'Option<[u8; 43]>', + 'rust_fmt': lambda x: None if x is None else Some(x), + }), + ('orchard_raw_addr', { + 'rust_type': 'Option<[u8; 43]>', + 'rust_fmt': lambda x: None if x is None else Some(x), + }), + ('unified_addr', 'Vec') + ), + test_vectors, + ) + + +if __name__ == "__main__": + main()