| 74 | |
| 75 | |
| 76 | class MiniWallet: |
| 77 | def __init__(self, test_node, *, mode=MiniWalletMode.ADDRESS_OP_TRUE, hrp="ert"): |
| 78 | self._test_node = test_node |
| 79 | self._utxos = [] |
| 80 | self._priv_key = None |
| 81 | self._address = None |
| 82 | |
| 83 | assert isinstance(mode, MiniWalletMode) |
| 84 | if mode == MiniWalletMode.RAW_OP_TRUE: |
| 85 | self._scriptPubKey = bytes(CScript([OP_TRUE])) |
| 86 | elif mode == MiniWalletMode.RAW_P2PK: |
| 87 | # use simple deterministic private key (k=1) |
| 88 | self._priv_key = ECKey() |
| 89 | self._priv_key.set((1).to_bytes(32, 'big'), True) |
| 90 | pub_key = self._priv_key.get_pubkey() |
| 91 | self._scriptPubKey = key_to_p2pk_script(pub_key.get_bytes()) |
| 92 | elif mode == MiniWalletMode.ADDRESS_OP_TRUE: |
| 93 | self._address, self._internal_key = create_deterministic_address_bcrt1_p2tr_op_true(hrp=hrp) |
| 94 | self._scriptPubKey = bytes.fromhex(self._test_node.validateaddress(self._address)['scriptPubKey']) |
| 95 | |
| 96 | def rescan_utxos(self): |
| 97 | """Drop all utxos and rescan the utxo set""" |
| 98 | self._utxos = [] |
| 99 | res = self._test_node.scantxoutset(action="start", scanobjects=[self.get_descriptor()]) |
| 100 | assert_equal(True, res['success']) |
| 101 | for utxo in res['unspents']: |
| 102 | self._utxos.append({'txid': utxo['txid'], 'vout': utxo['vout'], 'value': utxo['amount'], 'height': utxo['height']}) |
| 103 | |
| 104 | def scan_tx(self, tx): |
| 105 | """Scan the tx for self._scriptPubKey outputs and add them to self._utxos""" |
| 106 | for out in tx['vout']: |
| 107 | if out['scriptPubKey']['hex'] == self._scriptPubKey.hex(): |
| 108 | self._utxos.append({'txid': tx['txid'], 'vout': out['n'], 'value': out['value'], 'height': 0}) |
| 109 | |
| 110 | def sign_tx(self, tx, fixed_length=True): |
| 111 | """Sign tx that has been created by MiniWallet in P2PK mode""" |
| 112 | assert self._priv_key is not None |
| 113 | (sighash, err) = LegacySignatureHash(CScript(self._scriptPubKey), tx, 0, SIGHASH_ALL) |
| 114 | assert err is None |
| 115 | # for exact fee calculation, create only signatures with fixed size by default (>49.89% probability): |
| 116 | # 65 bytes: high-R val (33 bytes) + low-S val (32 bytes) |
| 117 | # with the DER header/skeleton data of 6 bytes added, this leads to a target size of 71 bytes |
| 118 | der_sig = b'' |
| 119 | while not len(der_sig) == 71: |
| 120 | der_sig = self._priv_key.sign_ecdsa(sighash) |
| 121 | if not fixed_length: |
| 122 | break |
| 123 | tx.vin[0].scriptSig = CScript([der_sig + bytes(bytearray([SIGHASH_ALL]))]) |
| 124 | |
| 125 | def generate(self, num_blocks, **kwargs): |
| 126 | """Generate blocks with coinbase outputs to the internal address, and append the outputs to the internal list""" |
| 127 | blocks = self._test_node.generatetodescriptor(num_blocks, self.get_descriptor(), **kwargs) |
| 128 | for b in blocks: |
| 129 | block_info = self._test_node.getblock(blockhash=b, verbosity=2) |
| 130 | cb_tx = block_info['tx'][0] |
| 131 | self._utxos.append({'txid': cb_tx['txid'], 'vout': 0, 'value': cb_tx['vout'][0]['value'], 'height': block_info['height']}) |
| 132 | return blocks |
| 133 |
no outgoing calls