if (m_print_to_file) {
        assert(!m_file_path.empty());
        m_fileout = fsbridge::fopen(m_file_path, "a");
        if (!m_fileout) {
            return false;
        }

        setbuf(m_fileout, nullptr); // unbuffered

            

Reported by FlawFinder.

                      // reopen the log file, if requested
        if (m_reopen_file) {
            m_reopen_file = false;
            FILE* new_fileout = fsbridge::fopen(m_file_path, "a");
            if (new_fileout) {
                setbuf(new_fileout, nullptr); // unbuffered
                fclose(m_fileout);
                m_fileout = new_fileout;
            }

            

Reported by FlawFinder.

                  assert(!m_file_path.empty());

    // Scroll debug.log if it's getting too big
    FILE* file = fsbridge::fopen(m_file_path, "r");

    // Special files (e.g. device nodes) may not have a size.
    size_t log_size = 0;
    try {
        log_size = fs::file_size(m_file_path);

            

Reported by FlawFinder.

                      int nBytes = fread(vch.data(), 1, vch.size(), file);
        fclose(file);

        file = fsbridge::fopen(m_file_path, "w");
        if (file)
        {
            fwrite(vch.data(), 1, nBytes, file);
            fclose(file);
        }

            

Reported by FlawFinder.

              
TYPED_TEST_SUITE_P(ExtendTest);

TYPED_TEST_P(ExtendTest, StandardResults) {
  // From rfc3720 section B.4.
  uint8_t buf[32];

  std::memset(buf, 0, sizeof(buf));
  EXPECT_EQ(static_cast<uint32_t>(0x8a9136aa),

            

Reported by Cppcheck.

              
TYPED_TEST_SUITE_P(ExtendTest);

TYPED_TEST_P(ExtendTest, StandardResults) {
  // From rfc3720 section B.4.
  uint8_t buf[32];

  std::memset(buf, 0, sizeof(buf));
  EXPECT_EQ(static_cast<uint32_t>(0x8a9136aa),

            

Reported by Cppcheck.

              
TYPED_TEST_SUITE_P(ExtendTest);

TYPED_TEST_P(ExtendTest, StandardResults) {
  // From rfc3720 section B.4.
  uint8_t buf[32];

  std::memset(buf, 0, sizeof(buf));
  EXPECT_EQ(static_cast<uint32_t>(0x8a9136aa),

            

Reported by Cppcheck.

              
TYPED_TEST_SUITE_P(ExtendTest);

TYPED_TEST_P(ExtendTest, StandardResults) {
  // From rfc3720 section B.4.
  uint8_t buf[32];

  std::memset(buf, 0, sizeof(buf));
  EXPECT_EQ(static_cast<uint32_t>(0x8a9136aa),

            

Reported by Cppcheck.

              from test_framework.test_framework import BitcoinTestFramework
from test_framework.util import assert_raises_rpc_error

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

    def skip_test_if_missing_module(self):

            

Reported by Pylint.

              
        # test sending a tx with disabled fallback fee (must fail)
        self.restart_node(0, extra_args=["-fallbackfee=0"])
        assert_raises_rpc_error(-6, "Fee estimation failed", lambda: self.nodes[0].sendtoaddress(self.nodes[0].getnewaddress(), 1))
        assert_raises_rpc_error(-4, "Fee estimation failed", lambda: self.nodes[0].fundrawtransaction(self.nodes[0].createrawtransaction([], {self.nodes[0].getnewaddress(): 1})))
        assert_raises_rpc_error(-6, "Fee estimation failed", lambda: self.nodes[0].sendmany("", {self.nodes[0].getnewaddress(): 1}))

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

            

Reported by Pylint.

                      # test sending a tx with disabled fallback fee (must fail)
        self.restart_node(0, extra_args=["-fallbackfee=0"])
        assert_raises_rpc_error(-6, "Fee estimation failed", lambda: self.nodes[0].sendtoaddress(self.nodes[0].getnewaddress(), 1))
        assert_raises_rpc_error(-4, "Fee estimation failed", lambda: self.nodes[0].fundrawtransaction(self.nodes[0].createrawtransaction([], {self.nodes[0].getnewaddress(): 1})))
        assert_raises_rpc_error(-6, "Fee estimation failed", lambda: self.nodes[0].sendmany("", {self.nodes[0].getnewaddress(): 1}))

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

            

Reported by Pylint.

                      self.restart_node(0, extra_args=["-fallbackfee=0"])
        assert_raises_rpc_error(-6, "Fee estimation failed", lambda: self.nodes[0].sendtoaddress(self.nodes[0].getnewaddress(), 1))
        assert_raises_rpc_error(-4, "Fee estimation failed", lambda: self.nodes[0].fundrawtransaction(self.nodes[0].createrawtransaction([], {self.nodes[0].getnewaddress(): 1})))
        assert_raises_rpc_error(-6, "Fee estimation failed", lambda: self.nodes[0].sendmany("", {self.nodes[0].getnewaddress(): 1}))

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

            

Reported by Pylint.

              # 2 hashes required per regtest block (with no difficulty adjustment)
REGTEST_WORK_PER_BLOCK = 2

class MinimumChainWorkTest(BitcoinTestFramework):
    def set_test_params(self):
        self.setup_clean_chain = True
        self.num_nodes = 3

        self.extra_args = [[], ["-minimumchainwork=0x65"], ["-minimumchainwork=0x65"]]

            

Reported by Pylint.

              
        starting_blockcount = self.nodes[2].getblockcount()

        num_blocks_to_generate = int((self.node_min_work[1] - starting_chain_work) / REGTEST_WORK_PER_BLOCK)
        self.log.info("Generating %d blocks on node0", num_blocks_to_generate)
        hashes = self.nodes[0].generatetoaddress(num_blocks_to_generate,
                                                 self.nodes[0].get_deterministic_priv_key().address)

        self.log.info("Node0 current chain work: %s", self.nodes[0].getblockheader(hashes[-1])['chainwork'])

            

Reported by Pylint.

                      hashes = self.nodes[0].generatetoaddress(num_blocks_to_generate,
                                                 self.nodes[0].get_deterministic_priv_key().address)

        self.log.info("Node0 current chain work: %s", self.nodes[0].getblockheader(hashes[-1])['chainwork'])

        # Sleep a few seconds and verify that node2 didn't get any new blocks
        # or headers.  We sleep, rather than sync_blocks(node0, node1) because
        # it's reasonable either way for node1 to get the blocks, or not get
        # them (since they're below node1's minchainwork).

            

Reported by Pylint.

                      assert_equal(len(self.nodes[2].getchaintips()), 1)
        assert_equal(self.nodes[2].getchaintips()[0]['height'], 0)

        assert self.nodes[1].getbestblockhash() != self.nodes[0].getbestblockhash()
        assert_equal(self.nodes[2].getblockcount(), starting_blockcount)

        self.log.info("Generating one more block")
        self.nodes[0].generatetoaddress(1, self.nodes[0].get_deterministic_priv_key().address)


            

Reported by Bandit.

              from test_framework.util import assert_equal


class ResendWalletTransactionsTest(BitcoinTestFramework):
    def set_test_params(self):
        self.num_nodes = 1

    def skip_test_if_missing_module(self):
        self.skip_if_no_wallet()

            

Reported by Pylint.

                      self.log.info("Create a block")
        # Create and submit a block without the transaction.
        # Transactions are only rebroadcast if there has been a block at least five minutes
        # after the last time we tried to broadcast. Use mocktime and give an extra minute to be sure.
        block_time = int(time.time()) + 6 * 60
        node.setmocktime(block_time)
        block = create_block(int(node.getbestblockhash(), 16), create_coinbase(node.getblockcount() + 1), block_time)
        block.rehash()
        block.solve()

            

Reported by Pylint.

                      # after the last time we tried to broadcast. Use mocktime and give an extra minute to be sure.
        block_time = int(time.time()) + 6 * 60
        node.setmocktime(block_time)
        block = create_block(int(node.getbestblockhash(), 16), create_coinbase(node.getblockcount() + 1), block_time)
        block.rehash()
        block.solve()
        node.submitblock(block.serialize().hex())

        # Set correct m_best_block_time, which is used in ResendWalletTransactions

            

Reported by Pylint.

                      self.log.info("Bump time & check that transaction is rebroadcast")
        # Transaction should be rebroadcast approximately 24 hours in the future,
        # but can range from 12-36. So bump 36 hours to be sure.
        with node.assert_debug_log(['ResendWalletTransactions: resubmit 1 unconfirmed transactions']):
            node.setmocktime(now + 36 * 60 * 60)
            # Tell scheduler to call MaybeResendWalletTxn now.
            node.mockscheduler(1)
        # Give some time for trickle to occur
        node.setmocktime(now + 36 * 60 * 60 + 600)

            

Reported by Pylint.

              from test_framework.test_framework import BitcoinTestFramework
from test_framework.util import assert_equal, mine_large_block

class TestP2PConn(P2PInterface):
    def __init__(self):
        super().__init__()
        self.block_receive_map = defaultdict(int)

    def on_inv(self, message):

            

Reported by Pylint.

                      message.block.calc_sha256()
        self.block_receive_map[message.block.sha256] += 1

class MaxUploadTest(BitcoinTestFramework):

    def set_test_params(self):
        self.setup_clean_chain = True
        self.num_nodes = 1
        self.extra_args = [[

            

Reported by Pylint.

                  def skip_test_if_missing_module(self):
        self.skip_if_no_wallet()

    def run_test(self):
        # Before we connect anything, we first set the time on the node
        # to be in the past, otherwise things break because the CNode
        # time counters can't be reset backward after initialization
        old_time = int(time.time() - 2*60*60*24*7)
        self.nodes[0].setmocktime(old_time)

            

Reported by Pylint.

                      getdata_request.inv = [CInv(MSG_BLOCK, big_old_block)]
        peer.send_and_ping(getdata_request)

        self.log.info("Peer still connected after trying to download old block (download permission)")
        peer_info = self.nodes[0].getpeerinfo()
        assert_equal(len(peer_info), 1)  # node is still connected
        assert_equal(peer_info[0]['permissions'], ['download'])



            

Reported by Pylint.

              from test_framework.wallet import MiniWallet


class MempoolSpendCoinbaseTest(BitcoinTestFramework):
    def set_test_params(self):
        self.num_nodes = 1

    def run_test(self):
        wallet = MiniWallet(self.nodes[0])

            

Reported by Pylint.

                  def run_test(self):
        wallet = MiniWallet(self.nodes[0])

        # Invalidate two blocks, so that miniwallet has access to a coin that will mature in the next block
        chain_height = 198
        self.nodes[0].invalidateblock(self.nodes[0].getblockhash(chain_height + 1))
        assert_equal(chain_height, self.nodes[0].getblockcount())

        # Coinbase at height chain_height-100+1 ok in mempool, should

            

Reported by Pylint.

                      wallet.scan_blocks(start=chain_height - 100 + 2, num=1)
        utxo_immature = wallet.get_utxo()

        spend_mature_id = wallet.send_self_transfer(from_node=self.nodes[0], utxo_to_spend=utxo_mature)["txid"]

        # other coinbase should be too immature to spend
        immature_tx = wallet.create_self_transfer(from_node=self.nodes[0], utxo_to_spend=utxo_immature, mempool_valid=False)
        assert_raises_rpc_error(-26,
                                "bad-txns-premature-spend-of-coinbase",

            

Reported by Pylint.

                      spend_mature_id = wallet.send_self_transfer(from_node=self.nodes[0], utxo_to_spend=utxo_mature)["txid"]

        # other coinbase should be too immature to spend
        immature_tx = wallet.create_self_transfer(from_node=self.nodes[0], utxo_to_spend=utxo_immature, mempool_valid=False)
        assert_raises_rpc_error(-26,
                                "bad-txns-premature-spend-of-coinbase",
                                lambda: self.nodes[0].sendrawtransaction(immature_tx['hex']))

        # mempool should have just the mature one

            

Reported by Pylint.

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

    const std::string& network = CBaseChainParams::MAIN;

            

Reported by FlawFinder.

              inline void WriteText(const fs::path& path, const std::string& text)
{
    fsbridge::ofstream file;
    file.open(path);
    file << text;
}

BOOST_FIXTURE_TEST_SUITE(settings_tests, BasicTestingSetup)


            

Reported by FlawFinder.

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

    const std::string& network = CBaseChainParams::MAIN;
    ForEachMergeSetup([&](const ActionList& arg_actions, const ActionList& conf_actions, bool force_set,

            

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.

                  ADDRS.append(addr)


class AddrReceiver(P2PInterface):
    addrv2_received_and_checked = False

    def __init__(self):
        super().__init__(support_addrv2 = True)


            

Reported by Pylint.

                      if expected_set == received_set:
            self.addrv2_received_and_checked = True

    def wait_for_addrv2(self):
        self.wait_until(lambda: "addrv2" in self.last_message)


class AddrTest(BitcoinTestFramework):
    def set_test_params(self):

            

Reported by Pylint.

                      self.wait_until(lambda: "addrv2" in self.last_message)


class AddrTest(BitcoinTestFramework):
    def set_test_params(self):
        self.setup_clean_chain = True
        self.num_nodes = 1
        self.extra_args = [["-whitelist=addr@127.0.0.1"]]


            

Reported by Pylint.

                          self.nodes[0].setmocktime(int(time.time()) + 30 * 60)
            addr_receiver.wait_for_addrv2()

        assert addr_receiver.addrv2_received_and_checked
        assert_equal(len(self.nodes[0].getnodeaddresses(count=0, network="i2p")), 0)


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

            

Reported by Bandit.

                      # Now relay again through nodes 3,4,5. Since 2 is outbound, it should remain HB.
        for nodeid in range(3, 6):
            status = self.relay_block_through(nodeid)
            assert status[0]
            assert status[nodeid - 2]
            assert_equal(sum(status), 3)

        # Reconnect peer 2, and retry. Now the three inbounds should be HB again.
        self.disconnect_nodes(1, 2)

            

Reported by Bandit.

                      for nodeid in range(3, 6):
            status = self.relay_block_through(nodeid)
            assert status[0]
            assert status[nodeid - 2]
            assert_equal(sum(status), 3)

        # Reconnect peer 2, and retry. Now the three inbounds should be HB again.
        self.disconnect_nodes(1, 2)
        self.connect_nodes(1, 2)

            

Reported by Bandit.

                      self.connect_nodes(1, 2)
        for nodeid in range(3, 6):
            status = self.relay_block_through(nodeid)
            assert not status[0]
            assert status[nodeid - 2]
        assert_equal(status, [False, True, True, True])


if __name__ == '__main__':

            

Reported by Bandit.

                      for nodeid in range(3, 6):
            status = self.relay_block_through(nodeid)
            assert not status[0]
            assert status[nodeid - 2]
        assert_equal(status, [False, True, True, True])


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

            

Reported by Bandit.