from decimal import Decimal

from test_framework.test_framework import BitcoinTestFramework
from test_framework.util import assert_equal, assert_greater_than, assert_raises_rpc_error, create_confirmed_utxos, create_lots_of_big_transactions, gen_return_txouts

class MempoolLimitTest(BitcoinTestFramework):
    def set_test_params(self):
        self.setup_clean_chain = True
        self.num_nodes = 1

            

Reported by Pylint.

              from test_framework.test_framework import BitcoinTestFramework
from test_framework.util import assert_equal, assert_greater_than, assert_raises_rpc_error, create_confirmed_utxos, create_lots_of_big_transactions, gen_return_txouts

class MempoolLimitTest(BitcoinTestFramework):
    def set_test_params(self):
        self.setup_clean_chain = True
        self.num_nodes = 1
        self.extra_args = [[
            "-acceptnonstdtxn=1",

            

Reported by Pylint.

                      us0 = utxos.pop()
        inputs = [{ "txid" : us0["txid"], "vout" : us0["vout"]}]
        outputs = {self.nodes[0].getnewaddress() : 0.0001}
        tx = self.nodes[0].createrawtransaction(inputs, outputs)
        self.nodes[0].settxfee(relayfee) # specifically fund this tx with low fee
        txF = self.nodes[0].fundrawtransaction(tx)
        self.nodes[0].settxfee(0) # return to automatic fee selection
        txFS = self.nodes[0].signrawtransactionwithwallet(txF['hex'])
        txid = self.nodes[0].sendrawtransaction(txFS['hex'])

            

Reported by Pylint.

                      outputs = {self.nodes[0].getnewaddress() : 0.0001}
        tx = self.nodes[0].createrawtransaction(inputs, outputs)
        self.nodes[0].settxfee(relayfee) # specifically fund this tx with low fee
        txF = self.nodes[0].fundrawtransaction(tx)
        self.nodes[0].settxfee(0) # return to automatic fee selection
        txFS = self.nodes[0].signrawtransactionwithwallet(txF['hex'])
        txid = self.nodes[0].sendrawtransaction(txFS['hex'])

        relayfee = self.nodes[0].getnetworkinfo()['relayfee']

            

Reported by Pylint.

                      self.nodes[0].settxfee(relayfee) # specifically fund this tx with low fee
        txF = self.nodes[0].fundrawtransaction(tx)
        self.nodes[0].settxfee(0) # return to automatic fee selection
        txFS = self.nodes[0].signrawtransactionwithwallet(txF['hex'])
        txid = self.nodes[0].sendrawtransaction(txFS['hex'])

        relayfee = self.nodes[0].getnetworkinfo()['relayfee']
        base_fee = relayfee*100
        for i in range (3):

            

Reported by Pylint.

                      base_fee = relayfee*100
        for i in range (3):
            txids.append([])
            txids[i] = create_lots_of_big_transactions(self.nodes[0], txouts, utxos[30*i:30*i+30], 30, (i+1)*base_fee)

        self.log.info('The tx should be evicted by now')
        assert txid not in self.nodes[0].getrawmempool()
        txdata = self.nodes[0].gettransaction(txid)
        assert txdata['confirmations'] ==  0  #confirmation should still be 0

            

Reported by Pylint.

                          txids[i] = create_lots_of_big_transactions(self.nodes[0], txouts, utxos[30*i:30*i+30], 30, (i+1)*base_fee)

        self.log.info('The tx should be evicted by now')
        assert txid not in self.nodes[0].getrawmempool()
        txdata = self.nodes[0].gettransaction(txid)
        assert txdata['confirmations'] ==  0  #confirmation should still be 0

        self.log.info('Check that mempoolminfee is larger than minrelytxfee')
        assert_equal(self.nodes[0].getmempoolinfo()['minrelaytxfee'], Decimal('0.00001000'))

            

Reported by Bandit.

                      self.log.info('The tx should be evicted by now')
        assert txid not in self.nodes[0].getrawmempool()
        txdata = self.nodes[0].gettransaction(txid)
        assert txdata['confirmations'] ==  0  #confirmation should still be 0

        self.log.info('Check that mempoolminfee is larger than minrelytxfee')
        assert_equal(self.nodes[0].getmempoolinfo()['minrelaytxfee'], Decimal('0.00001000'))
        assert_greater_than(self.nodes[0].getmempoolinfo()['mempoolminfee'], Decimal('0.00001000'))


            

Reported by Bandit.

                      us0 = utxos.pop()
        inputs = [{ "txid" : us0["txid"], "vout" : us0["vout"]}]
        outputs = {self.nodes[0].getnewaddress() : 0.0001}
        tx = self.nodes[0].createrawtransaction(inputs, outputs)
        # specifically fund this tx with a fee < mempoolminfee, >= than minrelaytxfee
        txF = self.nodes[0].fundrawtransaction(tx, {'feeRate': relayfee})
        txFS = self.nodes[0].signrawtransactionwithwallet(txF['hex'])
        assert_raises_rpc_error(-26, "mempool min fee not met", self.nodes[0].sendrawtransaction, txFS['hex'])


            

Reported by Pylint.

                      outputs = {self.nodes[0].getnewaddress() : 0.0001}
        tx = self.nodes[0].createrawtransaction(inputs, outputs)
        # specifically fund this tx with a fee < mempoolminfee, >= than minrelaytxfee
        txF = self.nodes[0].fundrawtransaction(tx, {'feeRate': relayfee})
        txFS = self.nodes[0].signrawtransactionwithwallet(txF['hex'])
        assert_raises_rpc_error(-26, "mempool min fee not met", self.nodes[0].sendrawtransaction, txFS['hex'])

if __name__ == '__main__':
    MempoolLimitTest().main()

            

Reported by Pylint.

              from test_framework.wallet import MiniWallet


class LongpollThread(threading.Thread):
    def __init__(self, node):
        threading.Thread.__init__(self)
        # query current longpollid
        template = node.getblocktemplate({'rules': ['segwit']})
        self.longpollid = template['longpollid']

            

Reported by Pylint.

                  def run(self):
        self.node.getblocktemplate({'longpollid': self.longpollid, 'rules': ['segwit']})

class GetBlockTemplateLPTest(BitcoinTestFramework):
    def set_test_params(self):
        self.num_nodes = 2
        self.supports_cli = False

    def run_test(self):

            

Reported by Pylint.

              
    def run_test(self):
        self.log.info("Warning: this test will take about 70 seconds in the best case. Be patient.")
        self.log.info("Test that longpollid doesn't change between successive getblocktemplate() invocations if nothing else happens")
        self.nodes[0].generate(10)
        template = self.nodes[0].getblocktemplate({'rules': ['segwit']})
        longpollid = template['longpollid']
        template2 = self.nodes[0].getblocktemplate({'rules': ['segwit']})
        assert template2['longpollid'] == longpollid

            

Reported by Pylint.

                      template = self.nodes[0].getblocktemplate({'rules': ['segwit']})
        longpollid = template['longpollid']
        template2 = self.nodes[0].getblocktemplate({'rules': ['segwit']})
        assert template2['longpollid'] == longpollid

        self.log.info("Test that longpoll waits if we do nothing")
        thr = LongpollThread(self.nodes[0])
        thr.start()
        # check that thread still lives

            

Reported by Bandit.

                      thr.start()
        # check that thread still lives
        thr.join(5)  # wait 5 seconds or until thread exits
        assert thr.is_alive()

        miniwallets = [ MiniWallet(node) for node in self.nodes ]
        self.log.info("Test that longpoll will terminate if another node generates a block")
        miniwallets[1].generate(1)  # generate a block on another node
        # check that thread will exit now that new transaction entered mempool

            

Reported by Bandit.

                      miniwallets[1].generate(1)  # generate a block on another node
        # check that thread will exit now that new transaction entered mempool
        thr.join(5)  # wait 5 seconds or until thread exits
        assert not thr.is_alive()

        self.log.info("Test that longpoll will terminate if we generate a block ourselves")
        thr = LongpollThread(self.nodes[0])
        thr.start()
        miniwallets[0].generate(1)  # generate a block on own node

            

Reported by Bandit.

                      thr.start()
        miniwallets[0].generate(1)  # generate a block on own node
        thr.join(5)  # wait 5 seconds or until thread exits
        assert not thr.is_alive()

        # Add enough mature utxos to the wallets, so that all txs spend confirmed coins
        self.nodes[0].generate(COINBASE_MATURITY)
        self.sync_blocks()


            

Reported by Bandit.

                      self.nodes[0].generate(COINBASE_MATURITY)
        self.sync_blocks()

        self.log.info("Test that introducing a new transaction into the mempool will terminate the longpoll")
        thr = LongpollThread(self.nodes[0])
        thr.start()
        # generate a random transaction and submit it
        min_relay_fee = self.nodes[0].getnetworkinfo()["relayfee"]
        fee_rate = min_relay_fee + Decimal('0.00000010') * random.randint(0,20)

            

Reported by Pylint.

                      thr.start()
        # generate a random transaction and submit it
        min_relay_fee = self.nodes[0].getnetworkinfo()["relayfee"]
        fee_rate = min_relay_fee + Decimal('0.00000010') * random.randint(0,20)
        miniwallets[0].send_self_transfer(from_node=random.choice(self.nodes),
                                          fee_rate=fee_rate)
        # after one minute, every 10 seconds the mempool is probed, so in 80 seconds it should have returned
        thr.join(60 + 20)
        assert not thr.is_alive()

            

Reported by Bandit.

                      # generate a random transaction and submit it
        min_relay_fee = self.nodes[0].getnetworkinfo()["relayfee"]
        fee_rate = min_relay_fee + Decimal('0.00000010') * random.randint(0,20)
        miniwallets[0].send_self_transfer(from_node=random.choice(self.nodes),
                                          fee_rate=fee_rate)
        # after one minute, every 10 seconds the mempool is probed, so in 80 seconds it should have returned
        thr.join(60 + 20)
        assert not thr.is_alive()


            

Reported by Bandit.

                      txid = node.sendrawtransaction(hexstring=tx_signed, maxfeerate=0)

        self.log.info("Test RPCResult scriptPubKey no longer returns the fields addresses or reqSigs by default")
        hash = node.generateblock(output=node.getnewaddress(), transactions=[txid])['hash']
        # Ensure both nodes have the newly generated block on disk.
        self.sync_blocks()
        script_pub_key = node.getblock(blockhash=hash, verbose=2)['tx'][-1]['vout'][0]['scriptPubKey']
        assert 'addresses' not in script_pub_key and 'reqSigs' not in script_pub_key


            

Reported by Pylint.

              )


class AddressesDeprecationTest(BitcoinTestFramework):
    def set_test_params(self):
        self.num_nodes = 2
        self.extra_args = [[], ["-deprecatedrpc=addresses"]]

    def skip_test_if_missing_module(self):

            

Reported by Pylint.

                  def run_test(self):
        self.test_addresses_deprecation()

    def test_addresses_deprecation(self):
        node = self.nodes[0]
        coin = node.listunspent().pop()

        inputs = [{'txid': coin['txid'], 'vout': coin['vout']}]
        outputs = {node.getnewaddress(): 0.99}

            

Reported by Pylint.

                      signed = node.signrawtransactionwithwallet(raw)['hex']

        # This transaction is derived from test/util/data/txcreatemultisig1.json
        tx = tx_from_hex(signed)
        tx.vout[0].scriptPubKey = bytes.fromhex("522102a5613bd857b7048924264d1e70e08fb2a7e6527d32b7ab1bb993ac59964ff39721021ac43c7ff740014c3b33737ede99c967e4764553d1b2b83db77c83b8715fa72d2102df2089105c77f266fa11a9d33f05c735234075f2e8780824c6b709415f9fb48553ae")
        tx_signed = node.signrawtransactionwithwallet(tx.serialize().hex())['hex']
        txid = node.sendrawtransaction(hexstring=tx_signed, maxfeerate=0)

        self.log.info("Test RPCResult scriptPubKey no longer returns the fields addresses or reqSigs by default")

            

Reported by Pylint.

              
        # This transaction is derived from test/util/data/txcreatemultisig1.json
        tx = tx_from_hex(signed)
        tx.vout[0].scriptPubKey = bytes.fromhex("522102a5613bd857b7048924264d1e70e08fb2a7e6527d32b7ab1bb993ac59964ff39721021ac43c7ff740014c3b33737ede99c967e4764553d1b2b83db77c83b8715fa72d2102df2089105c77f266fa11a9d33f05c735234075f2e8780824c6b709415f9fb48553ae")
        tx_signed = node.signrawtransactionwithwallet(tx.serialize().hex())['hex']
        txid = node.sendrawtransaction(hexstring=tx_signed, maxfeerate=0)

        self.log.info("Test RPCResult scriptPubKey no longer returns the fields addresses or reqSigs by default")
        hash = node.generateblock(output=node.getnewaddress(), transactions=[txid])['hash']

            

Reported by Pylint.

                      tx_signed = node.signrawtransactionwithwallet(tx.serialize().hex())['hex']
        txid = node.sendrawtransaction(hexstring=tx_signed, maxfeerate=0)

        self.log.info("Test RPCResult scriptPubKey no longer returns the fields addresses or reqSigs by default")
        hash = node.generateblock(output=node.getnewaddress(), transactions=[txid])['hash']
        # Ensure both nodes have the newly generated block on disk.
        self.sync_blocks()
        script_pub_key = node.getblock(blockhash=hash, verbose=2)['tx'][-1]['vout'][0]['scriptPubKey']
        assert 'addresses' not in script_pub_key and 'reqSigs' not in script_pub_key

            

Reported by Pylint.

                      hash = node.generateblock(output=node.getnewaddress(), transactions=[txid])['hash']
        # Ensure both nodes have the newly generated block on disk.
        self.sync_blocks()
        script_pub_key = node.getblock(blockhash=hash, verbose=2)['tx'][-1]['vout'][0]['scriptPubKey']
        assert 'addresses' not in script_pub_key and 'reqSigs' not in script_pub_key

        self.log.info("Test RPCResult scriptPubKey returns the addresses field with -deprecatedrpc=addresses")
        script_pub_key = self.nodes[1].getblock(blockhash=hash, verbose=2)['tx'][-1]['vout'][0]['scriptPubKey']
        assert_equal(script_pub_key['addresses'], ['mvKDK6D54HU8wQumJBLHY95eq5iHFqXSBz', 'mv3rHCQSwKp2BLSuMHD8uCS32LW5xiNAA5', 'mirrsyhAQYzo5CwVhcaYJKwUJu1WJRCRJe'])

            

Reported by Pylint.

                      # Ensure both nodes have the newly generated block on disk.
        self.sync_blocks()
        script_pub_key = node.getblock(blockhash=hash, verbose=2)['tx'][-1]['vout'][0]['scriptPubKey']
        assert 'addresses' not in script_pub_key and 'reqSigs' not in script_pub_key

        self.log.info("Test RPCResult scriptPubKey returns the addresses field with -deprecatedrpc=addresses")
        script_pub_key = self.nodes[1].getblock(blockhash=hash, verbose=2)['tx'][-1]['vout'][0]['scriptPubKey']
        assert_equal(script_pub_key['addresses'], ['mvKDK6D54HU8wQumJBLHY95eq5iHFqXSBz', 'mv3rHCQSwKp2BLSuMHD8uCS32LW5xiNAA5', 'mirrsyhAQYzo5CwVhcaYJKwUJu1WJRCRJe'])
        assert_equal(script_pub_key['reqSigs'], 2)

            

Reported by Bandit.

                      script_pub_key = node.getblock(blockhash=hash, verbose=2)['tx'][-1]['vout'][0]['scriptPubKey']
        assert 'addresses' not in script_pub_key and 'reqSigs' not in script_pub_key

        self.log.info("Test RPCResult scriptPubKey returns the addresses field with -deprecatedrpc=addresses")
        script_pub_key = self.nodes[1].getblock(blockhash=hash, verbose=2)['tx'][-1]['vout'][0]['scriptPubKey']
        assert_equal(script_pub_key['addresses'], ['mvKDK6D54HU8wQumJBLHY95eq5iHFqXSBz', 'mv3rHCQSwKp2BLSuMHD8uCS32LW5xiNAA5', 'mirrsyhAQYzo5CwVhcaYJKwUJu1WJRCRJe'])
        assert_equal(script_pub_key['reqSigs'], 2)



            

Reported by Pylint.

                      assert 'addresses' not in script_pub_key and 'reqSigs' not in script_pub_key

        self.log.info("Test RPCResult scriptPubKey returns the addresses field with -deprecatedrpc=addresses")
        script_pub_key = self.nodes[1].getblock(blockhash=hash, verbose=2)['tx'][-1]['vout'][0]['scriptPubKey']
        assert_equal(script_pub_key['addresses'], ['mvKDK6D54HU8wQumJBLHY95eq5iHFqXSBz', 'mv3rHCQSwKp2BLSuMHD8uCS32LW5xiNAA5', 'mirrsyhAQYzo5CwVhcaYJKwUJu1WJRCRJe'])
        assert_equal(script_pub_key['reqSigs'], 2)


if __name__ == "__main__":

            

Reported by Pylint.

              {
    CHash256 out_sha;
    FILE* out_file = nullptr;
    if (const char* out_path = getenv("ARGS_MERGE_TEST_OUT")) {
        out_file = fsbridge::fopen(out_path, "w");
        if (!out_file) throw std::system_error(errno, std::generic_category(), "fopen failed");
    }

    ForEachMergeSetup([&](const ActionList& arg_actions, const ActionList& conf_actions, bool soft_set, bool force_set,

            

Reported by FlawFinder.

              {
    CHash256 out_sha;
    FILE* out_file = nullptr;
    if (const char* out_path = getenv("CHAIN_MERGE_TEST_OUT")) {
        out_file = fsbridge::fopen(out_path, "w");
        if (!out_file) throw std::system_error(errno, std::generic_category(), "fopen failed");
    }

    ForEachMergeSetup([&](const ActionList& arg_actions, const ActionList& conf_actions) {

            

Reported by FlawFinder.

                  } while(0);
}

static const unsigned char ParseHex_expected[65] = {
    0x04, 0x67, 0x8a, 0xfd, 0xb0, 0xfe, 0x55, 0x48, 0x27, 0x19, 0x67, 0xf1, 0xa6, 0x71, 0x30, 0xb7,
    0x10, 0x5c, 0xd6, 0xa8, 0x28, 0xe0, 0x39, 0x09, 0xa6, 0x79, 0x62, 0xe0, 0xea, 0x1f, 0x61, 0xde,
    0xb6, 0x49, 0xf6, 0xbc, 0x3f, 0x4c, 0xef, 0x38, 0xc4, 0xf3, 0x55, 0x04, 0xe5, 0x1e, 0xc1, 0x12,
    0xde, 0x5c, 0x38, 0x4d, 0xf7, 0xba, 0x0b, 0x8d, 0x57, 0x8a, 0x4c, 0x70, 0x2b, 0x6b, 0xf1, 0x1d,
    0x5f

            

Reported by FlawFinder.

                  CHash256 out_sha;
    FILE* out_file = nullptr;
    if (const char* out_path = getenv("ARGS_MERGE_TEST_OUT")) {
        out_file = fsbridge::fopen(out_path, "w");
        if (!out_file) throw std::system_error(errno, std::generic_category(), "fopen failed");
    }

    ForEachMergeSetup([&](const ActionList& arg_actions, const ActionList& conf_actions, bool soft_set, bool force_set,
                          const std::string& section, const std::string& network, bool net_specific) {

            

Reported by FlawFinder.

                      out_file = nullptr;
    }

    unsigned char out_sha_bytes[CSHA256::OUTPUT_SIZE];
    out_sha.Finalize(out_sha_bytes);
    std::string out_sha_hex = HexStr(out_sha_bytes);

    // If check below fails, should manually dump the results with:
    //

            

Reported by FlawFinder.

                  CHash256 out_sha;
    FILE* out_file = nullptr;
    if (const char* out_path = getenv("CHAIN_MERGE_TEST_OUT")) {
        out_file = fsbridge::fopen(out_path, "w");
        if (!out_file) throw std::system_error(errno, std::generic_category(), "fopen failed");
    }

    ForEachMergeSetup([&](const ActionList& arg_actions, const ActionList& conf_actions) {
        TestArgsManager parser;

            

Reported by FlawFinder.

                      out_file = nullptr;
    }

    unsigned char out_sha_bytes[CSHA256::OUTPUT_SIZE];
    out_sha.Finalize(out_sha_bytes);
    std::string out_sha_hex = HexStr(out_sha_bytes);

    // If check below fails, should manually dump the results with:
    //

            

Reported by FlawFinder.

              {
    char ch;
    while (true) {
        int rv = read(fd, &ch, 1); // Wait for command
        assert(rv == 1);
        switch(ch) {
        case LockCommand:
            ch = LockDirectory(dirname, lockname);
            rv = write(fd, &ch, 1);

            

Reported by FlawFinder.

                  // Try to acquire lock in child process while we're holding it, this should fail.
    char ch;
    BOOST_CHECK_EQUAL(write(fd[1], &LockCommand, 1), 1);
    BOOST_CHECK_EQUAL(read(fd[1], &ch, 1), 1);
    BOOST_CHECK_EQUAL((bool)ch, false);

    // Give up our lock
    ReleaseDirectoryLocks();
    // Probing lock from our side now should succeed, but not hold on to the lock.

            

Reported by FlawFinder.

              
    // Try to acquire the lock in the child process, this should be successful.
    BOOST_CHECK_EQUAL(write(fd[1], &LockCommand, 1), 1);
    BOOST_CHECK_EQUAL(read(fd[1], &ch, 1), 1);
    BOOST_CHECK_EQUAL((bool)ch, true);

    // When we try to probe the lock now, it should fail.
    BOOST_CHECK_EQUAL(LockDirectory(dirname, lockname, true), false);


            

Reported by FlawFinder.

                  assert_raises_rpc_error,
)

class SignMessagesTest(BitcoinTestFramework):
    def set_test_params(self):
        self.setup_clean_chain = True
        self.num_nodes = 1
        self.extra_args = [["-addresstype=legacy"]]


            

Reported by Pylint.

                      self.log.info('test signing with priv_key')
        priv_key = 'cUeKHd5orzT3mz8P9pxyREHfsWtVfgsfDjiZZBcjUBAaGk1BTj7N'
        address = 'mpLQjfK79b7CCV4VMJWEWAj5Mpx8Up5zxB'
        expected_signature = 'INbVnW4e6PeRmsv2Qgu8NuopvrVjkcxob+sX8OcZG0SALhWybUjzMLPdAsXI46YZGb0KQTRii+wWIQzRpG/U+S0='
        signature = self.nodes[0].signmessagewithprivkey(priv_key, message)
        assert_equal(expected_signature, signature)
        assert self.nodes[0].verifymessage(address, signature, message)

        self.log.info('test signing with an address with wallet')

            

Reported by Pylint.

                      expected_signature = 'INbVnW4e6PeRmsv2Qgu8NuopvrVjkcxob+sX8OcZG0SALhWybUjzMLPdAsXI46YZGb0KQTRii+wWIQzRpG/U+S0='
        signature = self.nodes[0].signmessagewithprivkey(priv_key, message)
        assert_equal(expected_signature, signature)
        assert self.nodes[0].verifymessage(address, signature, message)

        self.log.info('test signing with an address with wallet')
        address = self.nodes[0].getnewaddress()
        signature = self.nodes[0].signmessage(address, message)
        assert self.nodes[0].verifymessage(address, signature, message)

            

Reported by Bandit.

                      self.log.info('test signing with an address with wallet')
        address = self.nodes[0].getnewaddress()
        signature = self.nodes[0].signmessage(address, message)
        assert self.nodes[0].verifymessage(address, signature, message)

        self.log.info('test verifying with another address should not work')
        other_address = self.nodes[0].getnewaddress()
        other_signature = self.nodes[0].signmessage(other_address, message)
        assert not self.nodes[0].verifymessage(other_address, signature, message)

            

Reported by Bandit.

                      self.log.info('test verifying with another address should not work')
        other_address = self.nodes[0].getnewaddress()
        other_signature = self.nodes[0].signmessage(other_address, message)
        assert not self.nodes[0].verifymessage(other_address, signature, message)
        assert not self.nodes[0].verifymessage(address, other_signature, message)

        self.log.info('test parameter validity and error codes')
        # signmessage(withprivkey) have two required parameters
        for num_params in [0, 1, 3, 4, 5]:

            

Reported by Bandit.

                      other_address = self.nodes[0].getnewaddress()
        other_signature = self.nodes[0].signmessage(other_address, message)
        assert not self.nodes[0].verifymessage(other_address, signature, message)
        assert not self.nodes[0].verifymessage(address, other_signature, message)

        self.log.info('test parameter validity and error codes')
        # signmessage(withprivkey) have two required parameters
        for num_params in [0, 1, 3, 4, 5]:
            param_list = ["dummy"]*num_params

            

Reported by Bandit.

                      # signmessage(withprivkey) have two required parameters
        for num_params in [0, 1, 3, 4, 5]:
            param_list = ["dummy"]*num_params
            assert_raises_rpc_error(-1, "signmessagewithprivkey", self.nodes[0].signmessagewithprivkey, *param_list)
            assert_raises_rpc_error(-1, "signmessage", self.nodes[0].signmessage, *param_list)
        # verifymessage has three required parameters
        for num_params in [0, 1, 2, 4, 5]:
            param_list = ["dummy"]*num_params
            assert_raises_rpc_error(-1, "verifymessage", self.nodes[0].verifymessage, *param_list)

            

Reported by Pylint.

                          param_list = ["dummy"]*num_params
            assert_raises_rpc_error(-1, "verifymessage", self.nodes[0].verifymessage, *param_list)
        # invalid key or address provided
        assert_raises_rpc_error(-5, "Invalid private key", self.nodes[0].signmessagewithprivkey, "invalid_key", message)
        assert_raises_rpc_error(-5, "Invalid address", self.nodes[0].signmessage, "invalid_addr", message)
        assert_raises_rpc_error(-5, "Invalid address", self.nodes[0].verifymessage, "invalid_addr", signature, message)
        # malformed signature provided
        assert_raises_rpc_error(-3, "Malformed base64 encoding", self.nodes[0].verifymessage, self.nodes[0].getnewaddress(), "invalid_sig", message)


            

Reported by Pylint.

                          assert_raises_rpc_error(-1, "verifymessage", self.nodes[0].verifymessage, *param_list)
        # invalid key or address provided
        assert_raises_rpc_error(-5, "Invalid private key", self.nodes[0].signmessagewithprivkey, "invalid_key", message)
        assert_raises_rpc_error(-5, "Invalid address", self.nodes[0].signmessage, "invalid_addr", message)
        assert_raises_rpc_error(-5, "Invalid address", self.nodes[0].verifymessage, "invalid_addr", signature, message)
        # malformed signature provided
        assert_raises_rpc_error(-3, "Malformed base64 encoding", self.nodes[0].verifymessage, self.nodes[0].getnewaddress(), "invalid_sig", message)

if __name__ == '__main__':

            

Reported by Pylint.

                      # invalid key or address provided
        assert_raises_rpc_error(-5, "Invalid private key", self.nodes[0].signmessagewithprivkey, "invalid_key", message)
        assert_raises_rpc_error(-5, "Invalid address", self.nodes[0].signmessage, "invalid_addr", message)
        assert_raises_rpc_error(-5, "Invalid address", self.nodes[0].verifymessage, "invalid_addr", signature, message)
        # malformed signature provided
        assert_raises_rpc_error(-3, "Malformed base64 encoding", self.nodes[0].verifymessage, self.nodes[0].getnewaddress(), "invalid_sig", message)

if __name__ == '__main__':
    SignMessagesTest().main()

            

Reported by Pylint.

                      self.num_nodes = 1

    def setup_network(self):
        self.alert_filename = os.path.join(self.options.tmpdir, "alert.txt")
        # Open and close to create zero-length file
        with open(self.alert_filename, 'w', encoding='utf8'):
            pass
        self.extra_args = [["-alertnotify=echo %s >> \"" + self.alert_filename + "\""]]
        self.setup_nodes()

            

Reported by Pylint.

                      # Open and close to create zero-length file
        with open(self.alert_filename, 'w', encoding='utf8'):
            pass
        self.extra_args = [["-alertnotify=echo %s >> \"" + self.alert_filename + "\""]]
        self.setup_nodes()

    def send_blocks_with_version(self, peer, numblocks, version):
        """Send numblocks blocks to peer with version set"""
        tip = self.nodes[0].getbestblockhash()

            

Reported by Pylint.

                      assert WARN_UNKNOWN_RULES_ACTIVE in node.getmininginfo()["warnings"]
        assert WARN_UNKNOWN_RULES_ACTIVE in node.getnetworkinfo()["warnings"]
        # Check that the alert file shows the versionbits unknown rules warning
        self.wait_until(lambda: self.versionbits_in_alert_file())

if __name__ == '__main__':
    VersionBitsWarningTest().main()

            

Reported by Pylint.

              WARN_UNKNOWN_RULES_ACTIVE = "Unknown new rules activated (versionbit {})".format(VB_UNKNOWN_BIT)
VB_PATTERN = re.compile("Unknown new rules activated.*versionbit")

class VersionBitsWarningTest(BitcoinTestFramework):
    def set_test_params(self):
        self.setup_clean_chain = True
        self.num_nodes = 1

    def setup_network(self):

            

Reported by Pylint.

                      # Mine one period worth of blocks
        node.generatetoaddress(VB_PERIOD, node_deterministic_address)

        self.log.info("Check that there is no warning if previous VB_BLOCKS have <VB_THRESHOLD blocks with unknown versionbits version.")
        # Build one period of blocks with < VB_THRESHOLD blocks signaling some unknown bit
        self.send_blocks_with_version(peer, VB_THRESHOLD - 1, VB_UNKNOWN_VERSION)
        node.generatetoaddress(VB_PERIOD - VB_THRESHOLD + 1, node_deterministic_address)

        # Check that we're not getting any versionbit-related errors in get*info()

            

Reported by Pylint.

                      node.generatetoaddress(VB_PERIOD - VB_THRESHOLD + 1, node_deterministic_address)

        # Check that we're not getting any versionbit-related errors in get*info()
        assert not VB_PATTERN.match(node.getmininginfo()["warnings"])
        assert not VB_PATTERN.match(node.getnetworkinfo()["warnings"])

        # Build one period of blocks with VB_THRESHOLD blocks signaling some unknown bit
        self.send_blocks_with_version(peer, VB_THRESHOLD, VB_UNKNOWN_VERSION)
        node.generatetoaddress(VB_PERIOD - VB_THRESHOLD, node_deterministic_address)

            

Reported by Bandit.

              
        # Check that we're not getting any versionbit-related errors in get*info()
        assert not VB_PATTERN.match(node.getmininginfo()["warnings"])
        assert not VB_PATTERN.match(node.getnetworkinfo()["warnings"])

        # Build one period of blocks with VB_THRESHOLD blocks signaling some unknown bit
        self.send_blocks_with_version(peer, VB_THRESHOLD, VB_UNKNOWN_VERSION)
        node.generatetoaddress(VB_PERIOD - VB_THRESHOLD, node_deterministic_address)


            

Reported by Bandit.

                      self.send_blocks_with_version(peer, VB_THRESHOLD, VB_UNKNOWN_VERSION)
        node.generatetoaddress(VB_PERIOD - VB_THRESHOLD, node_deterministic_address)

        self.log.info("Check that there is a warning if previous VB_BLOCKS have >=VB_THRESHOLD blocks with unknown versionbits version.")
        # Mine a period worth of expected blocks so the generic block-version warning
        # is cleared. This will move the versionbit state to ACTIVE.
        node.generatetoaddress(VB_PERIOD, node_deterministic_address)

        # Stop-start the node. This is required because bitcoind will only warn once about unknown versions or unknown rules activating.

            

Reported by Pylint.

                      # is cleared. This will move the versionbit state to ACTIVE.
        node.generatetoaddress(VB_PERIOD, node_deterministic_address)

        # Stop-start the node. This is required because bitcoind will only warn once about unknown versions or unknown rules activating.
        self.restart_node(0)

        # Generating one block guarantees that we'll get out of IBD
        node.generatetoaddress(1, node_deterministic_address)
        self.wait_until(lambda: not node.getblockchaininfo()['initialblockdownload'])

            

Reported by Pylint.

                      # Generating one more block will be enough to generate an error.
        node.generatetoaddress(1, node_deterministic_address)
        # Check that get*info() shows the versionbits unknown rules warning
        assert WARN_UNKNOWN_RULES_ACTIVE in node.getmininginfo()["warnings"]
        assert WARN_UNKNOWN_RULES_ACTIVE in node.getnetworkinfo()["warnings"]
        # Check that the alert file shows the versionbits unknown rules warning
        self.wait_until(lambda: self.versionbits_in_alert_file())

if __name__ == '__main__':

            

Reported by Bandit.

              uint16_t static inline ReadLE16(const unsigned char* ptr)
{
    uint16_t x;
    memcpy((char*)&x, ptr, 2);
    return le16toh(x);
}

uint32_t static inline ReadLE32(const unsigned char* ptr)
{

            

Reported by FlawFinder.

              uint32_t static inline ReadLE32(const unsigned char* ptr)
{
    uint32_t x;
    memcpy((char*)&x, ptr, 4);
    return le32toh(x);
}

uint64_t static inline ReadLE64(const unsigned char* ptr)
{

            

Reported by FlawFinder.

              uint64_t static inline ReadLE64(const unsigned char* ptr)
{
    uint64_t x;
    memcpy((char*)&x, ptr, 8);
    return le64toh(x);
}

void static inline WriteLE16(unsigned char* ptr, uint16_t x)
{

            

Reported by FlawFinder.

              void static inline WriteLE16(unsigned char* ptr, uint16_t x)
{
    uint16_t v = htole16(x);
    memcpy(ptr, (char*)&v, 2);
}

void static inline WriteLE32(unsigned char* ptr, uint32_t x)
{
    uint32_t v = htole32(x);

            

Reported by FlawFinder.

              void static inline WriteLE32(unsigned char* ptr, uint32_t x)
{
    uint32_t v = htole32(x);
    memcpy(ptr, (char*)&v, 4);
}

void static inline WriteLE64(unsigned char* ptr, uint64_t x)
{
    uint64_t v = htole64(x);

            

Reported by FlawFinder.

              void static inline WriteLE64(unsigned char* ptr, uint64_t x)
{
    uint64_t v = htole64(x);
    memcpy(ptr, (char*)&v, 8);
}

uint16_t static inline ReadBE16(const unsigned char* ptr)
{
    uint16_t x;

            

Reported by FlawFinder.

              uint16_t static inline ReadBE16(const unsigned char* ptr)
{
    uint16_t x;
    memcpy((char*)&x, ptr, 2);
    return be16toh(x);
}

uint32_t static inline ReadBE32(const unsigned char* ptr)
{

            

Reported by FlawFinder.

              uint32_t static inline ReadBE32(const unsigned char* ptr)
{
    uint32_t x;
    memcpy((char*)&x, ptr, 4);
    return be32toh(x);
}

uint64_t static inline ReadBE64(const unsigned char* ptr)
{

            

Reported by FlawFinder.

              uint64_t static inline ReadBE64(const unsigned char* ptr)
{
    uint64_t x;
    memcpy((char*)&x, ptr, 8);
    return be64toh(x);
}

void static inline WriteBE32(unsigned char* ptr, uint32_t x)
{

            

Reported by FlawFinder.

              void static inline WriteBE32(unsigned char* ptr, uint32_t x)
{
    uint32_t v = htobe32(x);
    memcpy(ptr, (char*)&v, 4);
}

void static inline WriteBE64(unsigned char* ptr, uint64_t x)
{
    uint64_t v = htobe64(x);

            

Reported by FlawFinder.

                      deps[module] = set()

# Iterate again, and build list of direct dependencies for each module
# TODO: implement support for multiple include directories
for arg in sorted(files.keys()):
    module = files[arg]
    with open(arg, 'r', encoding="utf8") as f:
        for line in f:
            match = RE.match(line)

            

Reported by Pylint.

              #!/usr/bin/env python3
# Copyright (c) 2018-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.

import sys
import re
from typing import Dict, List, Set


            

Reported by Pylint.

              #!/usr/bin/env python3
# Copyright (c) 2018-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.

import sys
import re
from typing import Dict, List, Set


            

Reported by Pylint.

                  'interfaces/'
]

def module_name(path):
    if path in MAPPING:
        path = MAPPING[path]
    if any(path.startswith(dirpath) for dirpath in HEADER_MODULE_PATHS):
        return path
    if path.endswith(".h"):

            

Reported by Pylint.

              ]

def module_name(path):
    if path in MAPPING:
        path = MAPPING[path]
    if any(path.startswith(dirpath) for dirpath in HEADER_MODULE_PATHS):
        return path
    if path.endswith(".h"):
        return path[:-2]

            

Reported by Pylint.

                          if match:
                include = match.group(1)
                included_module = module_name(include)
                if included_module is not None and included_module in deps and included_module != module:
                    deps[module].add(included_module)

# Loop to find the shortest (remaining) circular dependency
have_cycle: bool = False
while True:

            

Reported by Pylint.

              # Loop to find the shortest (remaining) circular dependency
have_cycle: bool = False
while True:
    shortest_cycle = None
    for module in sorted(deps.keys()):
        # Build the transitive closure of dependencies of module
        closure: Dict[str, List[str]] = dict()
        for dep in deps[module]:
            closure[dep] = []

            

Reported by Pylint.

                      for dep in deps[module]:
            closure[dep] = []
        while True:
            old_size = len(closure)
            old_closure_keys = sorted(closure.keys())
            for src in old_closure_keys:
                for dep in deps[src]:
                    if dep not in closure:
                        closure[dep] = closure[src] + [src]

            

Reported by Pylint.

                                      closure[dep] = closure[src] + [src]
            if len(closure) == old_size:
                break
        # If module is in its own transitive closure, it's a circular dependency; check if it is the shortest
        if module in closure and (shortest_cycle is None or len(closure[module]) + 1 < len(shortest_cycle)):
            shortest_cycle = [module] + closure[module]
    if shortest_cycle is None:
        break
    # We have the shortest circular dependency; report it

            

Reported by Pylint.

                          if len(closure) == old_size:
                break
        # If module is in its own transitive closure, it's a circular dependency; check if it is the shortest
        if module in closure and (shortest_cycle is None or len(closure[module]) + 1 < len(shortest_cycle)):
            shortest_cycle = [module] + closure[module]
    if shortest_cycle is None:
        break
    # We have the shortest circular dependency; report it
    module = shortest_cycle[0]

            

Reported by Pylint.

              
#include <boost/test/unit_test.hpp>

BOOST_FIXTURE_TEST_SUITE(crypto_tests, BasicTestingSetup)

template<typename Hasher, typename In, typename Out>
static void TestVector(const Hasher &h, const In &in, const Out &out) {
    Out hash;
    BOOST_CHECK(out.size() == h.OUTPUT_SIZE);

            

Reported by Cppcheck.

                  std::string info_stringified(reinterpret_cast<char*>(info.data()), info.size());

    CHKDF_HMAC_SHA256_L32 hkdf32(initial_key_material.data(), initial_key_material.size(), salt_stringified);
    unsigned char out[32];
    hkdf32.Expand32(info_stringified, out);
    BOOST_CHECK(HexStr(out) == okm_check_hex);
}

static std::string LongTestString()

            

Reported by FlawFinder.

              BOOST_AUTO_TEST_CASE(sha256d64)
{
    for (int i = 0; i <= 32; ++i) {
        unsigned char in[64 * 32];
        unsigned char out1[32 * 32], out2[32 * 32];
        for (int j = 0; j < 64 * i; ++j) {
            in[j] = InsecureRandBits(8);
        }
        for (int j = 0; j < i; ++j) {

            

Reported by FlawFinder.

              {
    for (int i = 0; i <= 32; ++i) {
        unsigned char in[64 * 32];
        unsigned char out1[32 * 32], out2[32 * 32];
        for (int j = 0; j < 64 * i; ++j) {
            in[j] = InsecureRandBits(8);
        }
        for (int j = 0; j < i; ++j) {
            CHash256().Write({in + 64 * j, 64}).Finalize({out1 + 32 * j, 32});

            

Reported by FlawFinder.

              
    SHA3_256 sha;
    // Hash the whole thing.
    unsigned char out[SHA3_256::OUTPUT_SIZE];
    sha.Write(in_bytes).Finalize(out);
    assert(out_bytes.size() == sizeof(out));
    BOOST_CHECK(std::equal(std::begin(out_bytes), std::end(out_bytes), out));

    // Reset and split randomly in 3

            

Reported by FlawFinder.

                  for (int i = 0; i < 262144; ++i) {
        KeccakF(state);
        for (int j = 0; j < 25; ++j) {
            unsigned char buf[8];
            WriteLE64(buf, state[j]);
            tester.Write(buf, 8);
        }
    }
    uint256 out;

            

Reported by FlawFinder.

              }

static MuHash3072 FromInt(unsigned char i) {
    unsigned char tmp[32] = {i, 0};
    return MuHash3072(tmp);
}

BOOST_AUTO_TEST_CASE(muhash_tests)
{

            

Reported by FlawFinder.

                  BOOST_CHECK_EQUAL(out, uint256S("10d312b100cbd32ada024a6646e40d3482fcff103668d2625f10002a607d5863"));

    MuHash3072 acc2 = FromInt(0);
    unsigned char tmp[32] = {1, 0};
    acc2.Insert(tmp);
    unsigned char tmp2[32] = {2, 0};
    acc2.Remove(tmp2);
    acc2.Finalize(out);
    BOOST_CHECK_EQUAL(out, uint256S("10d312b100cbd32ada024a6646e40d3482fcff103668d2625f10002a607d5863"));

            

Reported by FlawFinder.

                  MuHash3072 acc2 = FromInt(0);
    unsigned char tmp[32] = {1, 0};
    acc2.Insert(tmp);
    unsigned char tmp2[32] = {2, 0};
    acc2.Remove(tmp2);
    acc2.Finalize(out);
    BOOST_CHECK_EQUAL(out, uint256S("10d312b100cbd32ada024a6646e40d3482fcff103668d2625f10002a607d5863"));

    // Test MuHash3072 serialization

            

Reported by FlawFinder.

                  unsigned char out[SHA3_256::OUTPUT_SIZE];
    sha.Write(in_bytes).Finalize(out);
    assert(out_bytes.size() == sizeof(out));
    BOOST_CHECK(std::equal(std::begin(out_bytes), std::end(out_bytes), out));

    // Reset and split randomly in 3
    sha.Reset();
    int s1 = InsecureRandRange(in_bytes.size() + 1);
    int s2 = InsecureRandRange(in_bytes.size() + 1 - s1);

            

Reported by FlawFinder.

                      os.remove(os.path.join(node.cwd, "mock_result"))

    def run_test(self):
        self.log.debug(f"-signer={self.mock_signer_path()}")

        assert_raises_rpc_error(-1, 'Error: restart bitcoind with -signer=<cmd>',
            self.nodes[0].enumeratesigners
        )


            

Reported by Pylint.

              )


class RPCSignerTest(BitcoinTestFramework):
    def mock_signer_path(self):
        path = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'mocks', 'signer.py')
        if platform.system() == "Windows":
            return "py " + path
        else:

            

Reported by Pylint.

              

class RPCSignerTest(BitcoinTestFramework):
    def mock_signer_path(self):
        path = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'mocks', 'signer.py')
        if platform.system() == "Windows":
            return "py " + path
        else:
            return path

            

Reported by Pylint.

              

class RPCSignerTest(BitcoinTestFramework):
    def mock_signer_path(self):
        path = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'mocks', 'signer.py')
        if platform.system() == "Windows":
            return "py " + path
        else:
            return path

            

Reported by Pylint.

              class RPCSignerTest(BitcoinTestFramework):
    def mock_signer_path(self):
        path = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'mocks', 'signer.py')
        if platform.system() == "Windows":
            return "py " + path
        else:
            return path

    def set_test_params(self):

            

Reported by Pylint.

                  def skip_test_if_missing_module(self):
        self.skip_if_no_external_signer()

    def set_mock_result(self, node, res):
        with open(os.path.join(node.cwd, "mock_result"), "w", encoding="utf8") as f:
            f.write(res)

    def clear_mock_result(self, node):
        os.remove(os.path.join(node.cwd, "mock_result"))

            

Reported by Pylint.

                  def skip_test_if_missing_module(self):
        self.skip_if_no_external_signer()

    def set_mock_result(self, node, res):
        with open(os.path.join(node.cwd, "mock_result"), "w", encoding="utf8") as f:
            f.write(res)

    def clear_mock_result(self, node):
        os.remove(os.path.join(node.cwd, "mock_result"))

            

Reported by Pylint.

                      self.skip_if_no_external_signer()

    def set_mock_result(self, node, res):
        with open(os.path.join(node.cwd, "mock_result"), "w", encoding="utf8") as f:
            f.write(res)

    def clear_mock_result(self, node):
        os.remove(os.path.join(node.cwd, "mock_result"))


            

Reported by Pylint.

                      with open(os.path.join(node.cwd, "mock_result"), "w", encoding="utf8") as f:
            f.write(res)

    def clear_mock_result(self, node):
        os.remove(os.path.join(node.cwd, "mock_result"))

    def run_test(self):
        self.log.debug(f"-signer={self.mock_signer_path()}")


            

Reported by Pylint.

                      with open(os.path.join(node.cwd, "mock_result"), "w", encoding="utf8") as f:
            f.write(res)

    def clear_mock_result(self, node):
        os.remove(os.path.join(node.cwd, "mock_result"))

    def run_test(self):
        self.log.debug(f"-signer={self.mock_signer_path()}")


            

Reported by Pylint.