dash/test/functional/test_framework/blocktools.py

#!/usr/bin/env python3
# Copyright (c) 2015-2020 The Bitcoin Core developers
# Distributed under the MIT software license, see the accompanying
# file COPYING or http://www.opensource.org/licenses/mit-license.php.
"""Utilities for manipulating blocks and transactions."""

from binascii import a2b_hex
from decimal import Decimal
import io
import struct
import time
import unittest

from .messages import (
    CBlock,
    CCbTx,
    COIN,
    COutPoint,
    CTransaction,
    CTxIn,
    CTxOut,
    tx_from_hex,
    from_hex,
    uint256_to_string,
)
from .script import CScript, CScriptNum, CScriptOp, OP_TRUE, OP_CHECKSIG
from .util import assert_equal, hex_str_to_bytes
from io import BytesIO

MAX_BLOCK_SIGOPS = 40000

# Genesis block time (regtest)
TIME_GENESIS_BLOCK = 1417713337

# Coinbase transaction outputs can only be spent after this number of new blocks (network rule)
COINBASE_MATURITY = 100

NORMAL_GBT_REQUEST_PARAMS = {"rules": []} # type: ignore[var-annotated]

VERSIONBITS_LAST_OLD_BLOCK_VERSION = 4

def create_block(hashprev=None, coinbase=None, ntime=None, *, version=None, tmpl=None, txlist=None, dip4_activated=False, v20_activated=False):
    """Create a block (with regtest difficulty)."""
    block = CBlock()
    if tmpl is None:
        tmpl = {}
    block.nVersion = version or tmpl.get('version') or VERSIONBITS_LAST_OLD_BLOCK_VERSION
    block.nTime = ntime or tmpl.get('curtime') or int(time.time() + 600)
    block.hashPrevBlock = hashprev or int(tmpl['previousblockhash'], 0x10)
    if tmpl and not tmpl.get('bits') is None:
        block.nBits = struct.unpack('>I', a2b_hex(tmpl['bits']))[0]
    else:
        block.nBits = 0x207fffff  # difficulty retargeting is disabled in REGTEST chainparams
    if coinbase is None:
        coinbase = create_coinbase(height=tmpl['height'], dip4_activated=dip4_activated, v20_activated=v20_activated)
    block.vtx.append(coinbase)
    if txlist:
        for tx in txlist:
            if not hasattr(tx, 'calc_sha256'):
                txo = CTransaction()
                txo.deserialize(io.BytesIO(tx))
                tx = txo
            block.vtx.append(tx)
    block.hashMerkleRoot = block.calc_merkle_root()
    block.calc_sha256()
    return block

def create_block_with_mnpayments(mninfo, node, vtx=None, mn_payee=None, mn_amount=None):
    if vtx is None:
        vtx = []
    bt = node.getblocktemplate()
    height = bt['height']
    tip_hash = bt['previousblockhash']
    coinbasevalue = bt['coinbasevalue']

    assert len(bt['masternode']) <= 2
    if mn_payee is None:
        mn_payee = bt['masternode'][0]['payee']

    mn_operator_payee = None
    if len(bt['masternode']) == 2:
        mn_operator_payee = bt['masternode'][1]['payee']
    # we can't take the masternode payee amount from the template here as we might have additional fees in vtx

    # calculate fees that the block template included (we'll have to remove it from the coinbase as we won't
    # include the template's transactions
    bt_fees = 0
    for tx in bt['transactions']:
        bt_fees += tx['fee']

    new_fees = 0
    for tx in vtx:
        in_value = 0
        out_value = 0
        for txin in tx.vin:
            txout = node.gettxout(uint256_to_string(txin.prevout.hash), txin.prevout.n, False)
            in_value += int(txout['value'] * COIN)
        for txout in tx.vout:
            out_value += txout.nValue
        new_fees += in_value - out_value

    # fix fees
    coinbasevalue -= bt_fees
    coinbasevalue += new_fees

    operator_reward = 0
    if mn_operator_payee is not None:
        for mn in mninfo:
            if mn.rewards_address == mn_payee:
                operator_reward = mn.operator_reward
                break
        assert operator_reward > 0

    mn_operator_amount = 0
    if mn_amount is None:
        v20_info = node.getblockchaininfo()['softforks']['v20']
        mn_amount_total = get_masternode_payment(height, coinbasevalue, v20_info['active'])
        mn_operator_amount = mn_amount_total * operator_reward // 100
        mn_amount = mn_amount_total - mn_operator_amount
    miner_amount = coinbasevalue - mn_amount - mn_operator_amount

    miner_address = node.get_deterministic_priv_key().address
    outputs = {miner_address: str(Decimal(miner_amount) / COIN)}
    if mn_amount > 0:
        outputs[mn_payee] = str(Decimal(mn_amount) / COIN)
    if mn_operator_amount > 0:
        outputs[mn_operator_payee] = str(Decimal(mn_operator_amount) / COIN)

    coinbase = tx_from_hex(node.createrawtransaction([], outputs))
    coinbase.vin = create_coinbase(height).vin

    # We can't really use this one as it would result in invalid merkle roots for masternode lists
    if len(bt['coinbase_payload']) != 0:
        tip_block = node.getblock(tip_hash)
        cbtx = from_hex(CCbTx(version=1), bt['coinbase_payload'])
        if 'cbTx' in tip_block:
            cbtx.merkleRootMNList = int(tip_block['cbTx']['merkleRootMNList'], 16)
        else:
            cbtx.merkleRootMNList = 0
        coinbase.nVersion = 3
        coinbase.nType = 5 # CbTx
        coinbase.vExtraPayload = cbtx.serialize()

    coinbase.calc_sha256()

    block = create_block(int(tip_hash, 16), coinbase, ntime=bt['curtime'], version=bt['version'])
    block.vtx += vtx

    # Add quorum commitments from template
    for tx in bt['transactions']:
        tx2 = tx_from_hex(tx['data'])
        if tx2.nType == 6:
            block.vtx.append(tx2)

    block.hashMerkleRoot = block.calc_merkle_root()
    block.solve()
    return block

def script_BIP34_coinbase_height(height):
    if height <= 16:
        res = CScriptOp.encode_op_n(height)
        # Append dummy to increase scriptSig size above 2 (see bad-cb-length consensus rule)
        return CScript([res, OP_TRUE])
    return CScript([CScriptNum(height)])


def create_coinbase(height, pubkey=None, dip4_activated=False, v20_activated=False, nValue=500):
    """Create a coinbase transaction, assuming no miner fees.

    If pubkey is passed in, the coinbase output will be a P2PK output;
    otherwise an anyone-can-spend output."""
    coinbase = CTransaction()
    coinbase.vin.append(CTxIn(COutPoint(0, 0xffffffff), script_BIP34_coinbase_height(height), 0xffffffff))
    coinbaseoutput = CTxOut()
    coinbaseoutput.nValue = nValue * COIN
    if nValue == 500:
        halvings = int(height / 150)  # regtest
        coinbaseoutput.nValue >>= halvings
    if (pubkey is not None):
        coinbaseoutput.scriptPubKey = CScript([pubkey, OP_CHECKSIG])
    else:
        coinbaseoutput.scriptPubKey = CScript([OP_TRUE])
    coinbase.vout = [coinbaseoutput]
    if dip4_activated:
        coinbase.nVersion = 3
        coinbase.nType = 5
        cbtx_version = 3 if v20_activated else 2
        cbtx_payload = CCbTx(cbtx_version, height, 0, 0, 0)
        coinbase.vExtraPayload = cbtx_payload.serialize()
    coinbase.calc_sha256()
    return coinbase

def create_tx_with_script(prevtx, n, script_sig=b"", *, amount, script_pub_key=CScript()):
    """Return one-input, one-output transaction object
       spending the prevtx's n-th output with the given amount.

       Can optionally pass scriptPubKey and scriptSig, default is anyone-can-spend output.
    """
    tx = CTransaction()
    assert n < len(prevtx.vout)
    tx.vin.append(CTxIn(COutPoint(prevtx.sha256, n), script_sig, 0xffffffff))
    tx.vout.append(CTxOut(amount, script_pub_key))
    tx.calc_sha256()
    return tx

def create_transaction(node, txid, to_address, *, amount):
    """ Return signed transaction spending the first output of the
        input txid. Note that the node must have a wallet that can
        sign for the output that is being spent.
    """
    raw_tx = create_raw_transaction(node, txid, to_address, amount=amount)
    tx = CTransaction()
    tx.deserialize(BytesIO(hex_str_to_bytes(raw_tx)))
    return tx

def create_raw_transaction(node, txid, to_address, *, amount):
    """ Return raw signed transaction spending the first output of the
        input txid. Note that the node must have a wallet that can sign
        for the output that is being spent.
    """
    psbt = node.createpsbt(inputs=[{"txid": txid, "vout": 0}], outputs={to_address: amount})
    for _ in range(2):
        for w in node.listwallets():
            wrpc = node.get_wallet_rpc(w)
            signed_psbt = wrpc.walletprocesspsbt(psbt)
            psbt = signed_psbt['psbt']
    final_psbt = node.finalizepsbt(psbt)
    assert_equal(final_psbt["complete"], True)
    return final_psbt['hex']

def get_legacy_sigopcount_block(block, accurate=True):
    count = 0
    for tx in block.vtx:
        count += get_legacy_sigopcount_tx(tx, accurate)
    return count

def get_legacy_sigopcount_tx(tx, accurate=True):
    count = 0
    for i in tx.vout:
        count += i.scriptPubKey.GetSigOpCount(accurate)
    for j in tx.vin:
        # scriptSig might be of type bytes, so convert to CScript for the moment
        count += CScript(j.scriptSig).GetSigOpCount(accurate)
    return count

"""
Dash chaintips rpc returns extra info in each tip (difficulty, chainwork, and
forkpoint). Filter down to relevant ones checked in this test.
"""
def filter_tip_keys(chaintips):
    check_keys = ["hash", "height", "branchlen", "status"]
    filtered_tips = []
    for tip in chaintips:
        filtered_tips.append({k: tip[k] for k in check_keys})
    return filtered_tips

# Identical to GetMasternodePayment in C++ code
# TODO: remove it or make **proper** tests for various height
def get_masternode_payment(nHeight, blockValue, fV20Active):
    ret = int(blockValue / 5)

    nMNPIBlock = 350
    nMNPIPeriod = 10
    nReallocActivationHeight = 1

    if nHeight > nMNPIBlock:
        ret += int(blockValue / 20)
    if nHeight > nMNPIBlock+(nMNPIPeriod* 1):
        ret += int(blockValue / 20)
    if nHeight > nMNPIBlock+(nMNPIPeriod* 2):
        ret += int(blockValue / 20)
    if nHeight > nMNPIBlock+(nMNPIPeriod* 3):
        ret += int(blockValue / 40)
    if nHeight > nMNPIBlock+(nMNPIPeriod* 4):
        ret += int(blockValue / 40)
    if nHeight > nMNPIBlock+(nMNPIPeriod* 5):
        ret += int(blockValue / 40)
    if nHeight > nMNPIBlock+(nMNPIPeriod* 6):
        ret += int(blockValue / 40)
    if nHeight > nMNPIBlock+(nMNPIPeriod* 7):
        ret += int(blockValue / 40)
    if nHeight > nMNPIBlock+(nMNPIPeriod* 9):
        ret += int(blockValue / 40)

    if nHeight < nReallocActivationHeight:
        # Block Reward Realocation is not activated yet, nothing to do
        return ret

    nSuperblockCycle = 20
    # Actual realocation starts in the cycle next to one activation happens in
    nReallocStart = nReallocActivationHeight - nReallocActivationHeight % nSuperblockCycle + nSuperblockCycle

    if nHeight < nReallocStart:
        # Activated but we have to wait for the next cycle to start realocation, nothing to do
        return ret

    if fV20Active:
        # Once MNRewardReallocated activates, block reward is 80% of block subsidy (+ tx fees) since treasury is 20%
        # Since the MN reward needs to be equal to 60% of the block subsidy (according to the proposal), MN reward is set to 75% of the block reward.
        # Previous reallocation periods are dropped.
        return blockValue * 3 // 4

    # Periods used to reallocate the masternode reward from 50% to 60%
    vecPeriods = [
        513, # Period 1:  51.3%
        526, # Period 2:  52.6%
        533, # Period 3:  53.3%
        540, # Period 4:  54%
        546, # Period 5:  54.6%
        552, # Period 6:  55.2%
        557, # Period 7:  55.7%
        562, # Period 8:  56.2%
        567, # Period 9:  56.7%
        572, # Period 10: 57.2%
        577, # Period 11: 57.7%
        582, # Period 12: 58.2%
        585, # Period 13: 58.5%
        588, # Period 14: 58.8%
        591, # Period 15: 59.1%
        594, # Period 16: 59.4%
        597, # Period 17: 59.7%
        599, # Period 18: 59.9%
        600  # Period 19: 60%
    ]

    nReallocCycle = nSuperblockCycle * 3
    nCurrentPeriod = min(int((nHeight - nReallocStart) / nReallocCycle), len(vecPeriods) - 1)

    return int(blockValue * vecPeriods[nCurrentPeriod] / 1000)

class TestFrameworkBlockTools(unittest.TestCase):
    def test_create_coinbase(self):
        height = 20
        coinbase_tx = create_coinbase(height=height)
        assert_equal(CScriptNum.decode(coinbase_tx.vin[0].scriptSig), height)