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.

              
#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.

                  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.

                  time_t now = time(nullptr);
    fprintf(stderr, "Date:       %s", ctime(&now));  // ctime() adds newline

    FILE* cpuinfo = fopen("/proc/cpuinfo", "r");
    if (cpuinfo != nullptr) {
      char line[1000];
      int num_cpus = 0;
      std::string cpu_type;
      std::string cache_size;

            

Reported by FlawFinder.

              
    FILE* cpuinfo = fopen("/proc/cpuinfo", "r");
    if (cpuinfo != nullptr) {
      char line[1000];
      int num_cpus = 0;
      std::string cpu_type;
      std::string cache_size;
      while (fgets(line, sizeof(line), cpuinfo) != nullptr) {
        const char* sep = strchr(line, ':');

            

Reported by FlawFinder.

                  if (done_ < 1) done_ = 1;

    if (bytes_ > 0) {
      char rate[100];
      snprintf(rate, sizeof(rate), "%6.1f MB/s",
               (bytes_ / 1048576.0) / (finish - start_));
      if (!message_.empty()) {
        message_ = std::string(rate) + " " + message_;
      } else {

            

Reported by FlawFinder.

                  assert(db_ == nullptr);

    int status;
    char file_name[100];
    char* err_msg = nullptr;
    db_num_++;

    // Open database
    std::string tmp_dir;

            

Reported by FlawFinder.

                  }

    // Change SQLite cache size
    char cache_size[100];
    snprintf(cache_size, sizeof(cache_size), "PRAGMA cache_size = %d",
             FLAGS_num_pages);
    status = sqlite3_exec(db_, cache_size, nullptr, nullptr, &err_msg);
    ExecErrorCheck(status, err_msg);


            

Reported by FlawFinder.

              
    // FLAGS_page_size is defaulted to 1024
    if (FLAGS_page_size != 1024) {
      char page_size[100];
      snprintf(page_size, sizeof(page_size), "PRAGMA page_size = %d",
               FLAGS_page_size);
      status = sqlite3_exec(db_, page_size, nullptr, nullptr, &err_msg);
      ExecErrorCheck(status, err_msg);
    }

            

Reported by FlawFinder.

                  }

    if (num_entries != num_) {
      char msg[100];
      snprintf(msg, sizeof(msg), "(%d ops)", num_entries);
      message_ = msg;
    }

    char* err_msg = nullptr;

            

Reported by FlawFinder.

                      // Create values for key-value pair
        const int k =
            (order == SEQUENTIAL) ? i + j : (rand_.Next() % num_entries);
        char key[100];
        snprintf(key, sizeof(key), "%016d", k);

        // Bind KV values into replace_stmt
        status = sqlite3_bind_blob(replace_stmt, 1, key, 16, SQLITE_STATIC);
        ErrorCheck(status);

            

Reported by FlawFinder.

                    // Create and execute SQL statements
      for (int j = 0; j < entries_per_batch; j++) {
        // Create key value
        char key[100];
        int k = (order == SEQUENTIAL) ? i + j : (rand_.Next() % reads_);
        snprintf(key, sizeof(key), "%016d", k);

        // Bind key value into read_stmt
        status = sqlite3_bind_blob(read_stmt, 1, key, 16, SQLITE_STATIC);

            

Reported by FlawFinder.

                      ErrorCheck(status);

        // Execute replace_stmt
        bytes_ += value_size + strlen(key);
        status = sqlite3_step(replace_stmt);
        StepErrorCheck(status);

        // Reset SQLite statement for another use
        status = sqlite3_clear_bindings(replace_stmt);

            

Reported by FlawFinder.

                  if (script.size() == 25 && script[0] == OP_DUP && script[1] == OP_HASH160
                            && script[2] == 20 && script[23] == OP_EQUALVERIFY
                            && script[24] == OP_CHECKSIG) {
        memcpy(&hash, &script[3], 20);
        return true;
    }
    return false;
}


            

Reported by FlawFinder.

              {
    if (script.size() == 23 && script[0] == OP_HASH160 && script[1] == 20
                            && script[22] == OP_EQUAL) {
        memcpy(&hash, &script[2], 20);
        return true;
    }
    return false;
}


            

Reported by FlawFinder.

                  if (IsToKeyID(script, keyID)) {
        out.resize(21);
        out[0] = 0x00;
        memcpy(&out[1], &keyID, 20);
        return true;
    }
    CScriptID scriptID;
    if (IsToScriptID(script, scriptID)) {
        out.resize(21);

            

Reported by FlawFinder.

                  if (IsToScriptID(script, scriptID)) {
        out.resize(21);
        out[0] = 0x01;
        memcpy(&out[1], &scriptID, 20);
        return true;
    }
    CPubKey pubkey;
    if (IsToPubKey(script, pubkey)) {
        out.resize(33);

            

Reported by FlawFinder.

                  CPubKey pubkey;
    if (IsToPubKey(script, pubkey)) {
        out.resize(33);
        memcpy(&out[1], &pubkey[1], 32);
        if (pubkey[0] == 0x02 || pubkey[0] == 0x03) {
            out[0] = pubkey[0];
            return true;
        } else if (pubkey[0] == 0x04) {
            out[0] = 0x04 | (pubkey[64] & 0x01);

            

Reported by FlawFinder.

                      script[0] = OP_DUP;
        script[1] = OP_HASH160;
        script[2] = 20;
        memcpy(&script[3], in.data(), 20);
        script[23] = OP_EQUALVERIFY;
        script[24] = OP_CHECKSIG;
        return true;
    case 0x01:
        script.resize(23);

            

Reported by FlawFinder.

                      script.resize(23);
        script[0] = OP_HASH160;
        script[1] = 20;
        memcpy(&script[2], in.data(), 20);
        script[22] = OP_EQUAL;
        return true;
    case 0x02:
    case 0x03:
        script.resize(35);

            

Reported by FlawFinder.

                      script.resize(35);
        script[0] = 33;
        script[1] = nSize;
        memcpy(&script[2], in.data(), 32);
        script[34] = OP_CHECKSIG;
        return true;
    case 0x04:
    case 0x05:
        unsigned char vch[33] = {};

            

Reported by FlawFinder.

                      return true;
    case 0x04:
    case 0x05:
        unsigned char vch[33] = {};
        vch[0] = nSize - 2;
        memcpy(&vch[1], in.data(), 32);
        CPubKey pubkey{vch};
        if (!pubkey.Decompress())
            return false;

            

Reported by FlawFinder.

                  case 0x05:
        unsigned char vch[33] = {};
        vch[0] = nSize - 2;
        memcpy(&vch[1], in.data(), 32);
        CPubKey pubkey{vch};
        if (!pubkey.Decompress())
            return false;
        assert(pubkey.size() == 65);
        script.resize(67);

            

Reported by FlawFinder.

              
        for i, output_type in enumerate(["legacy", "p2sh-segwit", "bech32"]):

            self.log.info("Generate keys for wallet with address type: {}".format(output_type))
            idx = i+1
            for _ in range(90):
                addr_oldpool = self.nodes[idx].getnewaddress(address_type=output_type)
            for _ in range(20):
                addr_extpool = self.nodes[idx].getnewaddress(address_type=output_type)

            

Reported by Pylint.

              Two nodes. Node1 is under test. Node0 is providing transactions and generating blocks.

- Start node1, shutdown and backup wallet.
- Generate 110 keys (enough to drain the keypool). Store key 90 (in the initial keypool) and key 110 (beyond the initial keypool). Send funds to key 90 and key 110.
- Stop node1, clear the datadir, move wallet file back into the datadir and restart node1.
- connect node1 to node0. Verify that they sync and node1 receives its funds."""
import os
import shutil


            

Reported by Pylint.

              )


class KeypoolRestoreTest(BitcoinTestFramework):
    def set_test_params(self):
        self.setup_clean_chain = True
        self.num_nodes = 4
        self.extra_args = [[], ['-keypool=100'], ['-keypool=100'], ['-keypool=100']]


            

Reported by Pylint.

                      self.skip_if_no_wallet()

    def run_test(self):
        wallet_path = os.path.join(self.nodes[1].datadir, self.chain, "wallets", self.default_wallet_name, self.wallet_data_filename)
        wallet_backup_path = os.path.join(self.nodes[1].datadir, "wallet.bak")
        self.nodes[0].generate(COINBASE_MATURITY + 1)

        self.log.info("Make backup of wallet")
        self.stop_node(1)

            

Reported by Pylint.

                          # Make sure we're creating the outputs we expect
            address_details = self.nodes[idx].validateaddress(addr_extpool)
            if i == 0:
                assert not address_details["isscript"] and not address_details["iswitness"]
            elif i == 1:
                assert address_details["isscript"] and not address_details["iswitness"]
            else:
                assert not address_details["isscript"] and address_details["iswitness"]


            

Reported by Bandit.

                          if i == 0:
                assert not address_details["isscript"] and not address_details["iswitness"]
            elif i == 1:
                assert address_details["isscript"] and not address_details["iswitness"]
            else:
                assert not address_details["isscript"] and address_details["iswitness"]


            self.log.info("Send funds to wallet")

            

Reported by Bandit.

                          elif i == 1:
                assert address_details["isscript"] and not address_details["iswitness"]
            else:
                assert not address_details["isscript"] and address_details["iswitness"]


            self.log.info("Send funds to wallet")
            self.nodes[0].sendtoaddress(addr_oldpool, 10)
            self.nodes[0].generate(1)

            

Reported by Bandit.

                          # Check that we have marked all keys up to the used keypool key as used
            if self.options.descriptors:
                if output_type == 'legacy':
                    assert_equal(self.nodes[idx].getaddressinfo(self.nodes[idx].getnewaddress(address_type=output_type))['hdkeypath'], "m/44'/1'/0'/0/110")
                elif output_type == 'p2sh-segwit':
                    assert_equal(self.nodes[idx].getaddressinfo(self.nodes[idx].getnewaddress(address_type=output_type))['hdkeypath'], "m/49'/1'/0'/0/110")
                elif output_type == 'bech32':
                    assert_equal(self.nodes[idx].getaddressinfo(self.nodes[idx].getnewaddress(address_type=output_type))['hdkeypath'], "m/84'/1'/0'/0/110")
            else:

            

Reported by Pylint.

                              if output_type == 'legacy':
                    assert_equal(self.nodes[idx].getaddressinfo(self.nodes[idx].getnewaddress(address_type=output_type))['hdkeypath'], "m/44'/1'/0'/0/110")
                elif output_type == 'p2sh-segwit':
                    assert_equal(self.nodes[idx].getaddressinfo(self.nodes[idx].getnewaddress(address_type=output_type))['hdkeypath'], "m/49'/1'/0'/0/110")
                elif output_type == 'bech32':
                    assert_equal(self.nodes[idx].getaddressinfo(self.nodes[idx].getnewaddress(address_type=output_type))['hdkeypath'], "m/84'/1'/0'/0/110")
            else:
                assert_equal(self.nodes[idx].getaddressinfo(self.nodes[idx].getnewaddress(address_type=output_type))['hdkeypath'], "m/0'/0'/110'")


            

Reported by Pylint.

                              elif output_type == 'p2sh-segwit':
                    assert_equal(self.nodes[idx].getaddressinfo(self.nodes[idx].getnewaddress(address_type=output_type))['hdkeypath'], "m/49'/1'/0'/0/110")
                elif output_type == 'bech32':
                    assert_equal(self.nodes[idx].getaddressinfo(self.nodes[idx].getnewaddress(address_type=output_type))['hdkeypath'], "m/84'/1'/0'/0/110")
            else:
                assert_equal(self.nodes[idx].getaddressinfo(self.nodes[idx].getnewaddress(address_type=output_type))['hdkeypath'], "m/0'/0'/110'")


if __name__ == '__main__':

            

Reported by Pylint.

                  except ImportError:
        print("jinja2 not found. Try `pip install jinja2`")
        sys.exit(1)
    print(jinja2.Environment(loader=jinja2.FileSystemLoader('./'))
                    .get_template('combined_log_template.html')
                    .render(title="Combined Logs from testcase", log_events=[event._asdict() for event in log_events]))


if __name__ == '__main__':

            

Reported by Bandit.

                      'testdir', nargs='?', default='',
        help=('temporary test directory to combine logs from. '
              'Defaults to the most recent'))
    parser.add_argument('-c', '--color', dest='color', action='store_true', help='outputs the combined log with events colored by source (requires posix terminal colors. Use less -r for viewing)')
    parser.add_argument('--html', dest='html', action='store_true', help='outputs the combined log as html. Requires jinja2. pip install jinja2')
    args = parser.parse_args()

    if args.html and args.color:
        print("Only one out of --color or --html should be specified")

            

Reported by Pylint.

                      help=('temporary test directory to combine logs from. '
              'Defaults to the most recent'))
    parser.add_argument('-c', '--color', dest='color', action='store_true', help='outputs the combined log with events colored by source (requires posix terminal colors. Use less -r for viewing)')
    parser.add_argument('--html', dest='html', action='store_true', help='outputs the combined log as html. Requires jinja2. pip install jinja2')
    args = parser.parse_args()

    if args.html and args.color:
        print("Only one out of --color or --html should be specified")
        sys.exit(1)

            

Reported by Pylint.

                  glob = pathlib.Path(tmp_dir).glob('node0/**/debug.log')
    path = next(glob, None)
    if path:
        assert next(glob, None) is None #  more than one debug.log, should never happen
        chain = re.search(r'node0/(.+?)/debug\.log$', path.as_posix()).group(1)  # extract the chain name
    else:
        chain = 'regtest'  # fallback to regtest (should only happen when none exists)

    files = [("test", "%s/test_framework.log" % tmp_dir)]

            

Reported by Bandit.

                  path = next(glob, None)
    if path:
        assert next(glob, None) is None #  more than one debug.log, should never happen
        chain = re.search(r'node0/(.+?)/debug\.log$', path.as_posix()).group(1)  # extract the chain name
    else:
        chain = 'regtest'  # fallback to regtest (should only happen when none exists)

    files = [("test", "%s/test_framework.log" % tmp_dir)]
    for i in itertools.count():

            

Reported by Pylint.

                          if not os.path.isdir(folder):
                break
            for (_, _, fns) in os.walk(folder):
                for fn in fns:
                    warning = pathlib.Path('{}/{}'.format(folder, fn)).read_text().strip()
                    if warning:
                        warnings.append(("node{} {}".format(i, stream), warning))

    print()

            

Reported by Pylint.

                                      warnings.append(("node{} {}".format(i, stream), warning))

    print()
    for w in warnings:
        print("{} {} {} {}".format(colors[w[0].split()[0]], w[0], w[1], colors["reset"]))


def find_latest_test_dir():
    """Returns the latest tmpfile test directory prefix."""

            

Reported by Pylint.

                                  event = line
                # if it doesn't have a timestamp, it's a continuation line of the previous log.
                else:
                    # Add the line. Prefix with space equivalent to the source + timestamp so log lines are aligned
                    event += "                                   " + line
            # Flush the final event
            yield LogEvent(timestamp=timestamp, source=source, event=event.rstrip())
    except FileNotFoundError:
        print("File %s could not be opened. Continuing without it." % logfile, file=sys.stderr)

            

Reported by Pylint.

                  """Renders the iterator of log events into text."""
    for event in log_events:
        lines = event.event.splitlines()
        print("{0} {1: <5} {2} {3}".format(colors[event.source.rstrip()], event.source, lines[0], colors["reset"]))
        if len(lines) > 1:
            for line in lines[1:]:
                print("{0}{1}{2}".format(colors[event.source.rstrip()], line, colors["reset"]))



            

Reported by Pylint.

              def print_logs_html(log_events):
    """Renders the iterator of log events into html."""
    try:
        import jinja2
    except ImportError:
        print("jinja2 not found. Try `pip install jinja2`")
        sys.exit(1)
    print(jinja2.Environment(loader=jinja2.FileSystemLoader('./'))
                    .get_template('combined_log_template.html')

            

Reported by Pylint.