try:
            mallocinfo = node.getmemoryinfo(mode="mallocinfo")
            self.log.info('getmemoryinfo(mode="mallocinfo") call succeeded')
            tree = ET.fromstring(mallocinfo)
            assert_equal(tree.tag, 'malloc')
        except JSONRPCException:
            self.log.info('getmemoryinfo(mode="mallocinfo") not available')
            assert_raises_rpc_error(-8, 'mallocinfo is only available when compiled with glibc 2.10+', node.getmemoryinfo, mode="mallocinfo")


            

Reported by Bandit.

              # Distributed under the MIT software license, see the accompanying
# file COPYING or http://www.opensource.org/licenses/mit-license.php.
"""Test RPC misc output."""
import xml.etree.ElementTree as ET

from test_framework.test_framework import BitcoinTestFramework
from test_framework.util import (
    assert_raises_rpc_error,
    assert_equal,

            

Reported by Bandit.

              from test_framework.authproxy import JSONRPCException


class RpcMiscTest(BitcoinTestFramework):
    def set_test_params(self):
        self.num_nodes = 1
        self.supports_cli = False

    def run_test(self):

            

Reported by Pylint.

                          assert_equal(tree.tag, 'malloc')
        except JSONRPCException:
            self.log.info('getmemoryinfo(mode="mallocinfo") not available')
            assert_raises_rpc_error(-8, 'mallocinfo is only available when compiled with glibc 2.10+', node.getmemoryinfo, mode="mallocinfo")

        assert_raises_rpc_error(-8, "unknown mode foobar", node.getmemoryinfo, mode="foobar")

        self.log.info("test logging rpc and help")


            

Reported by Pylint.

                      # Test logging help returns the logging categories string in alphabetical order.
        categories = ', '.join(sorted_logging_categories)
        logging_help = self.nodes[0].help('logging')
        assert f"valid logging categories are: {categories}" in logging_help

        self.log.info("test echoipc (testing spawned process in multiprocess build)")
        assert_equal(node.echoipc("hello"), "hello")

        self.log.info("test getindexinfo")

            

Reported by Bandit.

                  std::vector<uint8_t> result(length);
    const std::vector<uint8_t> random_bytes = fuzzed_data_provider.ConsumeBytes<uint8_t>(length);
    if (!random_bytes.empty()) {
        std::memcpy(result.data(), random_bytes.data(), random_bytes.size());
    }
    return result;
}

CNetAddr ConsumeNetAddr(FuzzedDataProvider& fuzzed_data_provider) noexcept;

            

Reported by FlawFinder.

                  {
    }

    FILE* open();

    static ssize_t read(void* cookie, char* buf, size_t size);

    static ssize_t write(void* cookie, const char* buf, size_t size);


            

Reported by FlawFinder.

                  {
    }

    CAutoFile open()
    {
        return {m_fuzzed_file_provider.open(), m_fuzzed_data_provider.ConsumeIntegral<int>(), m_fuzzed_data_provider.ConsumeIntegral<int>()};
    }
};


            

Reported by FlawFinder.

              
    CAutoFile open()
    {
        return {m_fuzzed_file_provider.open(), m_fuzzed_data_provider.ConsumeIntegral<int>(), m_fuzzed_data_provider.ConsumeIntegral<int>()};
    }
};

[[nodiscard]] inline FuzzedAutoFileProvider ConsumeAutoFile(FuzzedDataProvider& fuzzed_data_provider) noexcept
{

            

Reported by FlawFinder.

              
    FILE* open();

    static ssize_t read(void* cookie, char* buf, size_t size);

    static ssize_t write(void* cookie, const char* buf, size_t size);

    static int seek(void* cookie, int64_t* offset, int whence);


            

Reported by FlawFinder.

              {
    CSipHasher hasher(0x0706050403020100ULL, 0x0F0E0D0C0B0A0908ULL);
    BOOST_CHECK_EQUAL(hasher.Finalize(),  0x726fdb47dd0e0e31ull);
    static const unsigned char t0[1] = {0};
    hasher.Write(t0, 1);
    BOOST_CHECK_EQUAL(hasher.Finalize(),  0x74f839c593dc67fdull);
    static const unsigned char t1[7] = {1,2,3,4,5,6,7};
    hasher.Write(t1, 7);
    BOOST_CHECK_EQUAL(hasher.Finalize(),  0x93f5f5799a932462ull);

            

Reported by FlawFinder.

                  static const unsigned char t0[1] = {0};
    hasher.Write(t0, 1);
    BOOST_CHECK_EQUAL(hasher.Finalize(),  0x74f839c593dc67fdull);
    static const unsigned char t1[7] = {1,2,3,4,5,6,7};
    hasher.Write(t1, 7);
    BOOST_CHECK_EQUAL(hasher.Finalize(),  0x93f5f5799a932462ull);
    hasher.Write(0x0F0E0D0C0B0A0908ULL);
    BOOST_CHECK_EQUAL(hasher.Finalize(),  0x3f2acc7f57c29bdbull);
    static const unsigned char t2[2] = {16,17};

            

Reported by FlawFinder.

                  BOOST_CHECK_EQUAL(hasher.Finalize(),  0x93f5f5799a932462ull);
    hasher.Write(0x0F0E0D0C0B0A0908ULL);
    BOOST_CHECK_EQUAL(hasher.Finalize(),  0x3f2acc7f57c29bdbull);
    static const unsigned char t2[2] = {16,17};
    hasher.Write(t2, 2);
    BOOST_CHECK_EQUAL(hasher.Finalize(),  0x4bc1b3f0968dd39cull);
    static const unsigned char t3[9] = {18,19,20,21,22,23,24,25,26};
    hasher.Write(t3, 9);
    BOOST_CHECK_EQUAL(hasher.Finalize(),  0x2f2e6163076bcfadull);

            

Reported by FlawFinder.

                  static const unsigned char t2[2] = {16,17};
    hasher.Write(t2, 2);
    BOOST_CHECK_EQUAL(hasher.Finalize(),  0x4bc1b3f0968dd39cull);
    static const unsigned char t3[9] = {18,19,20,21,22,23,24,25,26};
    hasher.Write(t3, 9);
    BOOST_CHECK_EQUAL(hasher.Finalize(),  0x2f2e6163076bcfadull);
    static const unsigned char t4[5] = {27,28,29,30,31};
    hasher.Write(t4, 5);
    BOOST_CHECK_EQUAL(hasher.Finalize(),  0x7127512f72f27cceull);

            

Reported by FlawFinder.

                  static const unsigned char t3[9] = {18,19,20,21,22,23,24,25,26};
    hasher.Write(t3, 9);
    BOOST_CHECK_EQUAL(hasher.Finalize(),  0x2f2e6163076bcfadull);
    static const unsigned char t4[5] = {27,28,29,30,31};
    hasher.Write(t4, 5);
    BOOST_CHECK_EQUAL(hasher.Finalize(),  0x7127512f72f27cceull);
    hasher.Write(0x2726252423222120ULL);
    BOOST_CHECK_EQUAL(hasher.Finalize(),  0x0e3ea96b5304a7d0ull);
    hasher.Write(0x2F2E2D2C2B2A2928ULL);

            

Reported by FlawFinder.

              NORMAL_FEE_FILTER = Decimal(100) / COIN


class P2PIBDTxRelayTest(BitcoinTestFramework):
    def set_test_params(self):
        self.setup_clean_chain = True
        self.num_nodes = 2
        self.extra_args = [
            ["-minrelaytxfee={}".format(NORMAL_FEE_FILTER)],

            

Reported by Pylint.

                  def run_test(self):
        self.log.info("Check that nodes set minfilter to MAX_MONEY while still in IBD")
        for node in self.nodes:
            assert node.getblockchaininfo()['initialblockdownload']
            self.wait_until(lambda: all(peer['minfeefilter'] == MAX_FEE_FILTER for peer in node.getpeerinfo()))

        # Come out of IBD by generating a block
        self.nodes[0].generate(1)
        self.sync_all()

            

Reported by Bandit.

                      self.log.info("Check that nodes set minfilter to MAX_MONEY while still in IBD")
        for node in self.nodes:
            assert node.getblockchaininfo()['initialblockdownload']
            self.wait_until(lambda: all(peer['minfeefilter'] == MAX_FEE_FILTER for peer in node.getpeerinfo()))

        # Come out of IBD by generating a block
        self.nodes[0].generate(1)
        self.sync_all()


            

Reported by Pylint.

              
        self.log.info("Check that nodes reset minfilter after coming out of IBD")
        for node in self.nodes:
            assert not node.getblockchaininfo()['initialblockdownload']
            self.wait_until(lambda: all(peer['minfeefilter'] == NORMAL_FEE_FILTER for peer in node.getpeerinfo()))


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

            

Reported by Bandit.

                      self.log.info("Check that nodes reset minfilter after coming out of IBD")
        for node in self.nodes:
            assert not node.getblockchaininfo()['initialblockdownload']
            self.wait_until(lambda: all(peer['minfeefilter'] == NORMAL_FEE_FILTER for peer in node.getpeerinfo()))


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

            

Reported by Pylint.

                      except AssertionError as e:
            raised = True
            if not re.search(r"Expected messages .* does not partially match log", str(e)):
                raise AssertionError(f"Assertion raised as expected, but with an unexpected message: {str(e)}")
        if not raised:
            raise AssertionError("Assertion should have been raised")

        self.log.info("Ensure we try to connect if port=0 and get an error due to missing I2P proxy")
        addr = "h3r6bkn46qxftwja53pxiykntegfyfjqtnzbm6iv6r5mungmqgmq.b32.i2p:0"

            

Reported by Pylint.

              from test_framework.test_framework import BitcoinTestFramework


class I2PPorts(BitcoinTestFramework):
    def set_test_params(self):
        self.num_nodes = 1
        # The test assumes that an I2P SAM proxy is not listening here.
        self.extra_args = [["-i2psam=127.0.0.1:60000"]]


            

Reported by Pylint.

                      try:
            with node.assert_debug_log(expected_msgs=[f"Error connecting to {addr}"]):
                node.addnode(node=addr, command="onetry")
        except AssertionError as e:
            raised = True
            if not re.search(r"Expected messages .* does not partially match log", str(e)):
                raise AssertionError(f"Assertion raised as expected, but with an unexpected message: {str(e)}")
        if not raised:
            raise AssertionError("Assertion should have been raised")

            

Reported by Pylint.

                      except AssertionError as e:
            raised = True
            if not re.search(r"Expected messages .* does not partially match log", str(e)):
                raise AssertionError(f"Assertion raised as expected, but with an unexpected message: {str(e)}")
        if not raised:
            raise AssertionError("Assertion should have been raised")

        self.log.info("Ensure we try to connect if port=0 and get an error due to missing I2P proxy")
        addr = "h3r6bkn46qxftwja53pxiykntegfyfjqtnzbm6iv6r5mungmqgmq.b32.i2p:0"

            

Reported by Pylint.

                      if not raised:
            raise AssertionError("Assertion should have been raised")

        self.log.info("Ensure we try to connect if port=0 and get an error due to missing I2P proxy")
        addr = "h3r6bkn46qxftwja53pxiykntegfyfjqtnzbm6iv6r5mungmqgmq.b32.i2p:0"
        with node.assert_debug_log(expected_msgs=[f"Error connecting to {addr}"]):
            node.addnode(node=addr, command="onetry")



            

Reported by Pylint.

                Env* env_;
};

TEST(MemEnvTest, Basics) {
  uint64_t file_size;
  WritableFile* writable_file;
  std::vector<std::string> children;

  ASSERT_OK(env_->CreateDir("/dir"));

            

Reported by Cppcheck.

                SequentialFile* seq_file;
  RandomAccessFile* rand_file;
  Slice result;
  char scratch[100];

  ASSERT_OK(env_->CreateDir("/dir"));

  ASSERT_OK(env_->NewWritableFile("/dir/f", &writable_file));
  ASSERT_OK(writable_file->Append("hello "));

            

Reported by FlawFinder.

                // Verify that overwriting an open file will result in the new file data
  // being read from files opened before the write.
  Slice result;
  char scratch[kFileDataLen];
  ASSERT_OK(rand_file->Read(0, kFileDataLen, &result, scratch));
  ASSERT_EQ(0, result.compare(kWrite2Data));

  delete rand_file;
}

            

Reported by FlawFinder.

              
  size_t read = 0;
  std::string read_data;
  while (read < kWriteSize) {
    ASSERT_OK(seq_file->Read(kWriteSize - read, &result, scratch));
    read_data.append(result.data(), result.size());
    read += result.size();
  }
  ASSERT_TRUE(write_data == read_data);

            

Reported by FlawFinder.

                size_t read = 0;
  std::string read_data;
  while (read < kWriteSize) {
    ASSERT_OK(seq_file->Read(kWriteSize - read, &result, scratch));
    read_data.append(result.data(), result.size());
    read += result.size();
  }
  ASSERT_TRUE(write_data == read_data);
  delete seq_file;

            

Reported by FlawFinder.

              };

// Test empty table/block.
TEST(Harness, Empty) {
  for (int i = 0; i < kNumTestArgs; i++) {
    Init(kTestArgList[i]);
    Random rnd(test::RandomSeed() + 1);
    Test(&rnd);
  }

            

Reported by Cppcheck.

                  if (offset + n > contents_.size()) {
      n = contents_.size() - offset;
    }
    memcpy(scratch, &contents_[offset], n);
    *result = Slice(scratch, n);
    return Status::OK();
  }

  std::string GetName() const { return ""; }

            

Reported by FlawFinder.

              // code never generates such blocks, but the Java version of leveldb
// seems to.
TEST(Harness, ZeroRestartPointsInBlock) {
  char data[sizeof(uint32_t)];
  memset(data, 0, sizeof(data));
  BlockContents contents;
  contents.data = Slice(data, sizeof(data));
  contents.cachable = false;
  contents.heap_allocated = false;

            

Reported by FlawFinder.

                int files = 0;
  for (int level = 0; level < config::kNumLevels; level++) {
    std::string value;
    char name[100];
    snprintf(name, sizeof(name), "leveldb.num-files-at-level%d", level);
    ASSERT_TRUE(db()->GetProperty(name, &value));
    files += atoi(value.c_str());
  }
  ASSERT_GT(files, 0);

            

Reported by FlawFinder.

                  char name[100];
    snprintf(name, sizeof(name), "leveldb.num-files-at-level%d", level);
    ASSERT_TRUE(db()->GetProperty(name, &value));
    files += atoi(value.c_str());
  }
  ASSERT_GT(files, 0);
}

class MemTableTest {};

            

Reported by FlawFinder.

              from test_framework.util import assert_equal


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

    def run_test(self):

            

Reported by Pylint.

              
        self.log.info("test -uacomment max length")
        self.stop_node(0)
        expected = r"Error: Total length of network version string \([0-9]+\) exceeds maximum length \(256\). Reduce the number or size of uacomments."
        self.nodes[0].assert_start_raises_init_error(["-uacomment=" + 'a' * 256], expected, match=ErrorMatch.FULL_REGEX)

        self.log.info("test -uacomment unsafe characters")
        for unsafe_char in ['/', ':', '(', ')', '₿', '🏃']:
            expected = r"Error: User Agent comment \(" + re.escape(unsafe_char) + r"\) contains unsafe characters."

            

Reported by Pylint.

                      self.log.info("test -uacomment max length")
        self.stop_node(0)
        expected = r"Error: Total length of network version string \([0-9]+\) exceeds maximum length \(256\). Reduce the number or size of uacomments."
        self.nodes[0].assert_start_raises_init_error(["-uacomment=" + 'a' * 256], expected, match=ErrorMatch.FULL_REGEX)

        self.log.info("test -uacomment unsafe characters")
        for unsafe_char in ['/', ':', '(', ')', '₿', '🏃']:
            expected = r"Error: User Agent comment \(" + re.escape(unsafe_char) + r"\) contains unsafe characters."
            self.nodes[0].assert_start_raises_init_error(["-uacomment=" + unsafe_char], expected, match=ErrorMatch.FULL_REGEX)

            

Reported by Pylint.

              
        self.log.info("test -uacomment unsafe characters")
        for unsafe_char in ['/', ':', '(', ')', '₿', '🏃']:
            expected = r"Error: User Agent comment \(" + re.escape(unsafe_char) + r"\) contains unsafe characters."
            self.nodes[0].assert_start_raises_init_error(["-uacomment=" + unsafe_char], expected, match=ErrorMatch.FULL_REGEX)


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

            

Reported by Pylint.

                      self.log.info("test -uacomment unsafe characters")
        for unsafe_char in ['/', ':', '(', ')', '₿', '🏃']:
            expected = r"Error: User Agent comment \(" + re.escape(unsafe_char) + r"\) contains unsafe characters."
            self.nodes[0].assert_start_raises_init_error(["-uacomment=" + unsafe_char], expected, match=ErrorMatch.FULL_REGEX)


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

            

Reported by Pylint.

                  softfork_active,
)

class SegwitUpgradeTest(BitcoinTestFramework):
    def set_test_params(self):
        self.setup_clean_chain = True
        self.num_nodes = 1
        self.extra_args = [["-segwitheight=10"]]


            

Reported by Pylint.

                      # Node hasn't been used or connected yet
        assert_equal(node.getblockcount(), 0)

        assert not softfork_active(node, "segwit")

        # Generate 8 blocks without witness data
        node.generate(8)
        assert_equal(node.getblockcount(), 8)


            

Reported by Bandit.

              
        self.stop_node(0)
        # Restarting the node (with segwit activation height set to 5) should result in a shutdown
        # because the blockchain consists of 3 insufficiently validated blocks per segwit consensus rules.
        node.assert_start_raises_init_error(
                extra_args=["-segwitheight=5"],
                expected_msg=": Witness data for blocks after height 5 requires validation. Please restart with -reindex..\nPlease restart with -reindex or -reindex-chainstate to recover.")

        # As directed, the user restarts the node with -reindex

            

Reported by Pylint.

                      # because the blockchain consists of 3 insufficiently validated blocks per segwit consensus rules.
        node.assert_start_raises_init_error(
                extra_args=["-segwitheight=5"],
                expected_msg=": Witness data for blocks after height 5 requires validation. Please restart with -reindex..\nPlease restart with -reindex or -reindex-chainstate to recover.")

        # As directed, the user restarts the node with -reindex
        self.start_node(0, extra_args=["-reindex", "-segwitheight=5"])

        # With the segwit consensus rules, the node is able to validate only up to block 4

            

Reported by Pylint.

                      assert_equal(node.getblockcount(), 4)

        # The upgraded node should now have segwit activated
        assert softfork_active(node, "segwit")


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

            

Reported by Bandit.

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

from argparse import ArgumentParser
from base64 import urlsafe_b64encode
from binascii import hexlify
from getpass import getpass

            

Reported by Pylint.

                  """Create 32 byte b64 password"""
    return urlsafe_b64encode(urandom(32)).decode('utf-8')

def password_to_hmac(salt, password):
    m = hmac.new(bytearray(salt, 'utf-8'), bytearray(password, 'utf-8'), 'SHA256')
    return m.hexdigest()

def main():
    parser = ArgumentParser(description='Create login credentials for a JSON-RPC user')

            

Reported by Pylint.

                  return urlsafe_b64encode(urandom(32)).decode('utf-8')

def password_to_hmac(salt, password):
    m = hmac.new(bytearray(salt, 'utf-8'), bytearray(password, 'utf-8'), 'SHA256')
    return m.hexdigest()

def main():
    parser = ArgumentParser(description='Create login credentials for a JSON-RPC user')
    parser.add_argument('username', help='the username for authentication')

            

Reported by Pylint.

                  m = hmac.new(bytearray(salt, 'utf-8'), bytearray(password, 'utf-8'), 'SHA256')
    return m.hexdigest()

def main():
    parser = ArgumentParser(description='Create login credentials for a JSON-RPC user')
    parser.add_argument('username', help='the username for authentication')
    parser.add_argument('password', help='leave empty to generate a random password or specify "-" to prompt for password', nargs='?')
    args = parser.parse_args()


            

Reported by Pylint.

              def main():
    parser = ArgumentParser(description='Create login credentials for a JSON-RPC user')
    parser.add_argument('username', help='the username for authentication')
    parser.add_argument('password', help='leave empty to generate a random password or specify "-" to prompt for password', nargs='?')
    args = parser.parse_args()

    if not args.password:
        args.password = generate_password()
    elif args.password == '-':

            

Reported by Pylint.