Inside Bitcoin

251        else:
252            utxo_filter = self._utxos
253        if confirmed_only:
254            utxo_filter = filter(lambda utxo: utxo['confirmations'] > 0, utxo_filter)
255        utxos = deepcopy(list(utxo_filter))
256        if mark_as_spent:
257            self._utxos = []
258        return utxos
259
260    def send_self_transfer(self, *, from_node, **kwargs):
261        """Call create_self_transfer and send the transaction."""
262        tx = self.create_self_transfer(**kwargs)
263        self.sendrawtransaction(from_node=from_node, tx_hex=tx['hex'])
264        return tx
265
266    def send_to(self, *, from_node, scriptPubKey, amount, fee=1000):
267        """
268        Create and send a tx with an output to a given scriptPubKey/amount,
269        plus a change output to our internal address. To keep things simple, a
270        fixed fee given in Satoshi is used.
271
272        Note that this method fails if there is no single internal utxo
273        available that can cover the cost for the amount and the fixed fee
274        (the utxo with the largest value is taken).
275        """
276        tx = self.create_self_transfer(fee_rate=0)["tx"]
277        assert_greater_than_or_equal(tx.vout[0].nValue, amount + fee)
278        tx.vout[0].nValue -= (amount + fee)           # change output -> MiniWallet
279        tx.vout.append(CTxOut(amount, scriptPubKey))  # arbitrary output -> to be returned
280        txid = self.sendrawtransaction(from_node=from_node, tx_hex=tx.serialize().hex())
281        return {
282            "sent_vout": 1,
283            "txid": txid,
284            "wtxid": tx.wtxid_hex,
285            "hex": tx.serialize().hex(),
286            "tx": tx,
287        }
288
289    def send_self_transfer_multi(self, *, from_node, **kwargs):
290        """Call create_self_transfer_multi and send the transaction."""
291        tx = self.create_self_transfer_multi(**kwargs)
292        self.sendrawtransaction(from_node=from_node, tx_hex=tx["hex"])
293        return tx
294
295    def create_self_transfer_multi(
296        self,
297        *,
298        utxos_to_spend: Optional[list[dict]] = None,
299        num_outputs=1,
300        amount_per_output=0,
301        version=2,
302        locktime=0,
303        sequence=0,
304        fee_per_output=1000,
305        target_vsize=0,
306        confirmed_only=False,
307    ):
308        """
309        Create and return a transaction that spends the given UTXOs and creates a
310        certain number of outputs with equal amounts. The output amounts can be
311        set by amount_per_output or automatically calculated with a fee_per_output.
312        """
313        utxos_to_spend = utxos_to_spend or [self.get_utxo(confirmed_only=confirmed_only)]
314        sequence = [sequence] * len(utxos_to_spend) if type(sequence) is int else sequence
315        assert_equal(len(utxos_to_spend), len(sequence))
316
317        # calculate output amount
318        inputs_value_total = sum([int(COIN * utxo['value']) for utxo in utxos_to_spend])
319        outputs_value_total = inputs_value_total - fee_per_output * num_outputs
320        amount_per_output = amount_per_output or (outputs_value_total // num_outputs)
321        assert amount_per_output > 0
322        outputs_value_total = amount_per_output * num_outputs
323        fee = Decimal(inputs_value_total - outputs_value_total) / COIN
324
325        # create tx
326        tx = CTransaction()
327        tx.vin = [CTxIn(COutPoint(int(utxo_to_spend['txid'], 16), utxo_to_spend['vout']), nSequence=seq) for utxo_to_spend, seq in zip(utxos_to_spend, sequence)]
328        tx.vout = [CTxOut(amount_per_output, bytearray(self._scriptPubKey)) for _ in range(num_outputs)]
329        tx.version = version
330        tx.nLockTime = locktime
331
332        self.sign_tx(tx)
333
334        if target_vsize:
335            self._bulk_tx(tx, target_vsize)
336
337        txid = tx.txid_hex
338        return {
339            "new_utxos": [self._create_utxo(
340                txid=txid,
341                vout=i,
342                value=Decimal(tx.vout[i].nValue) / COIN,
343                height=0,
344                coinbase=False,
345                confirmations=0,
346            ) for i in range(len(tx.vout))],
347            "fee": fee,
348            "txid": txid,
349            "wtxid": tx.wtxid_hex,
350            "hex": tx.serialize().hex(),
351            "tx": tx,
352        }
353
354    def create_self_transfer(
355            self,
356            *,
357            fee_rate=Decimal("0.003"),
358            fee=Decimal("0"),
359            utxo_to_spend=None,
360            target_vsize=0,
361            confirmed_only=False,
362            **kwargs,
363    ):
364        """Create and return a tx with the specified fee. If fee is 0, use fee_rate, where the resulting fee may be exact or at most one satoshi higher than needed."""
365        utxo_to_spend = utxo_to_spend or self.get_utxo(confirmed_only=confirmed_only)
366        assert fee_rate >= 0
367        assert fee >= 0
368        # calculate fee
369        if self._mode in (MiniWalletMode.RAW_OP_TRUE, MiniWalletMode.ADDRESS_OP_TRUE):
370            vsize = Decimal(104)  # anyone-can-spend
371        elif self._mode == MiniWalletMode.RAW_P2PK:
372            vsize = Decimal(168)  # P2PK (73 bytes scriptSig + 35 bytes scriptPubKey + 60 bytes other)
373        else:
374            assert False
375        if target_vsize and not fee:  # respect fee_rate if target vsize is passed
376            fee = get_fee(target_vsize, fee_rate)
377        send_value = utxo_to_spend["value"] - (fee or (fee_rate * vsize / 1000))
378        if send_value <= 0:
379            raise RuntimeError(f"UTXO value {utxo_to_spend['value']} is too small to cover fees {(fee or (fee_rate * vsize / 1000))}")
380        # create tx
381        tx = self.create_self_transfer_multi(
382            utxos_to_spend=[utxo_to_spend],
383            amount_per_output=int(COIN * send_value),
384            target_vsize=target_vsize,
385            **kwargs,
386        )
387        if not target_vsize:
388            assert_equal(tx["tx"].get_vsize(), vsize)
389        tx["new_utxo"] = tx.pop("new_utxos")[0]
390
391        return tx
392
393    def sendrawtransaction(self, *, from_node, tx_hex, maxfeerate=0, **kwargs):
394        txid = from_node.sendrawtransaction(hexstring=tx_hex, maxfeerate=maxfeerate, **kwargs)
395        self.scan_tx(from_node.decoderawtransaction(tx_hex))
396        return txid
397
398    def create_self_transfer_chain(self, *, chain_length, utxo_to_spend=None):
399        """
400        Create a "chain" of chain_length transactions. The nth transaction in
401        the chain is a child of the n-1th transaction and parent of the n+1th transaction.
402        """
403        chaintip_utxo = utxo_to_spend or self.get_utxo()
404        chain = []
405
406        for _ in range(chain_length):
407            tx = self.create_self_transfer(utxo_to_spend=chaintip_utxo)
408            chaintip_utxo = tx["new_utxo"]
409            chain.append(tx)
410
411        return chain
412
413    def send_self_transfer_chain(self, *, from_node, **kwargs):
414        """Create and send a "chain" of chain_length transactions. The nth transaction in
415        the chain is a child of the n-1th transaction and parent of the n+1th transaction.
416
417        Returns a list of objects for each tx (see create_self_transfer_multi).
418        """
419        chain = self.create_self_transfer_chain(**kwargs)
420        for t in chain:
421            self.sendrawtransaction(from_node=from_node, tx_hex=t["hex"])
422        return chain
423
424
425class NodeSigner:
426    """Simple wallet replacement that delegates signing of existing raw transactions to a node by
427       using the `signrawtransactionwithkey` RPC. This can be used for spending from widespread
428       output types (P2PKH, P2WPKH, P2SH-P2WPKH, P2TR) without having the wallet compiled in."""
429    def __init__(self, node):
430        self._node = node
431        self._key_entries = []
432
433    def getnewaddress(self, address_type='legacy'):
434        (seckey, pubkey), spk, address = getnewdestination(address_type)
435        redeem_script = key_to_p2wpkh_script(pubkey) if address_type == 'p2sh-segwit' else None
436        self._key_entries.append({"seckey_wif": bytes_to_wif(seckey.get_bytes()), "output_script": spk, "redeem_script": redeem_script})
437        return pubkey, spk, address
438
439    def listunspent(self):
440        needles = [descsum_create(f'raw({key_entry["output_script"].hex()})') for key_entry in self._key_entries]
441        scan_res = self._node.scantxoutset(action="start", scanobjects=needles)
442        spend_height = scan_res['height'] + 1  # coins would be spent in the next block
443        unspents = []
444        for u in scan_res['unspents']:
445            if u["coinbase"] and (spend_height - u["height"]) < COINBASE_MATURITY:  # skip immature coins
446                continue
447            unspent = { "txid": u["txid"], "vout": u["vout"], "scriptPubKey": u["scriptPubKey"], "amount": u["amount"] }
448            key_entry = [ke for ke in self._key_entries if ke["output_script"] == bytes.fromhex(u["scriptPubKey"])][0]
449            if key_entry["redeem_script"] is not None:
450                unspent["redeemScript"] = key_entry["redeem_script"].hex()
451            unspents.append(unspent)
452        return unspents
453
454    def signrawtransaction(self, tx_hex, inputs):
455        output_scripts_to_sign = {i["scriptPubKey"] for i in inputs}
456        seckeys_wif = [ke["seckey_wif"] for ke in self._key_entries if ke["output_script"].hex() in output_scripts_to_sign]
457        return self._node.signrawtransactionwithkey(tx_hex, seckeys_wif, inputs)
458
459
460def getnewdestination(address_type='bech32m'):
461    """Generate a random destination of the specified type and return the
462       corresponding key pair, scriptPubKey and address. Supported types are
463       'legacy', 'p2sh-segwit', 'bech32' and 'bech32m'. Can be used when a random
464       destination is needed, but no compiled wallet is available (e.g. as
465       replacement to the getnewaddress/getaddressinfo RPCs)."""
466    key, pubkey = generate_keypair()
467    if address_type == 'legacy':
468        scriptpubkey = key_to_p2pkh_script(pubkey)
469        address = key_to_p2pkh(pubkey)
470    elif address_type == 'p2sh-segwit':
471        scriptpubkey = key_to_p2sh_p2wpkh_script(pubkey)
472        address = key_to_p2sh_p2wpkh(pubkey)
473    elif address_type == 'bech32':
474        scriptpubkey = key_to_p2wpkh_script(pubkey)
475        address = key_to_p2wpkh(pubkey)
476    elif address_type == 'bech32m':
477        tap = taproot_construct(compute_xonly_pubkey(key.get_bytes())[0])
478        pubkey = tap.output_pubkey
479        scriptpubkey = tap.scriptPubKey
480        address = output_key_to_p2tr(pubkey)
481    else:
482        assert False
483    return (key, pubkey), scriptpubkey, address
484