# file COPYING or http://www.opensource.org/licenses/mit-license.php.
"""Test per-peer message capture capability.

Additionally, the output of contrib/message-capture/message-capture-parser.py should be verified manually.
"""

import glob
from io import BytesIO
import os

            

Reported by Pylint.

                  """
    with open(dat_file, 'rb') as f_in:
        # This should have at least one message in it
        assert(os.fstat(f_in.fileno()).st_size >= TIME_SIZE + LENGTH_SIZE + MSGTYPE_SIZE)
        while True:
            tmp_header_raw = f_in.read(TIME_SIZE + LENGTH_SIZE + MSGTYPE_SIZE)
            if not tmp_header_raw:
                break
            tmp_header = BytesIO(tmp_header_raw)

            

Reported by Bandit.

                  """
    with open(dat_file, 'rb') as f_in:
        # This should have at least one message in it
        assert(os.fstat(f_in.fileno()).st_size >= TIME_SIZE + LENGTH_SIZE + MSGTYPE_SIZE)
        while True:
            tmp_header_raw = f_in.read(TIME_SIZE + LENGTH_SIZE + MSGTYPE_SIZE)
            if not tmp_header_raw:
                break
            tmp_header = BytesIO(tmp_header_raw)

            

Reported by Pylint.

                          raw_msgtype = tmp_header.read(MSGTYPE_SIZE)
            msgtype: bytes = raw_msgtype.split(b'\x00', 1)[0]
            remainder =  raw_msgtype.split(b'\x00', 1)[1]
            assert(len(msgtype) > 0)
            assert(msgtype in MESSAGEMAP)
            assert(len(remainder) == 0 or not remainder.decode().isprintable())
            length: int = int.from_bytes(tmp_header.read(LENGTH_SIZE), "little")
            data = f_in.read(length)
            assert_equal(len(data), length)

            

Reported by Bandit.

                          raw_msgtype = tmp_header.read(MSGTYPE_SIZE)
            msgtype: bytes = raw_msgtype.split(b'\x00', 1)[0]
            remainder =  raw_msgtype.split(b'\x00', 1)[1]
            assert(len(msgtype) > 0)
            assert(msgtype in MESSAGEMAP)
            assert(len(remainder) == 0 or not remainder.decode().isprintable())
            length: int = int.from_bytes(tmp_header.read(LENGTH_SIZE), "little")
            data = f_in.read(length)
            assert_equal(len(data), length)

            

Reported by Pylint.

                          msgtype: bytes = raw_msgtype.split(b'\x00', 1)[0]
            remainder =  raw_msgtype.split(b'\x00', 1)[1]
            assert(len(msgtype) > 0)
            assert(msgtype in MESSAGEMAP)
            assert(len(remainder) == 0 or not remainder.decode().isprintable())
            length: int = int.from_bytes(tmp_header.read(LENGTH_SIZE), "little")
            data = f_in.read(length)
            assert_equal(len(data), length)


            

Reported by Bandit.

                          msgtype: bytes = raw_msgtype.split(b'\x00', 1)[0]
            remainder =  raw_msgtype.split(b'\x00', 1)[1]
            assert(len(msgtype) > 0)
            assert(msgtype in MESSAGEMAP)
            assert(len(remainder) == 0 or not remainder.decode().isprintable())
            length: int = int.from_bytes(tmp_header.read(LENGTH_SIZE), "little")
            data = f_in.read(length)
            assert_equal(len(data), length)


            

Reported by Pylint.

                          remainder =  raw_msgtype.split(b'\x00', 1)[1]
            assert(len(msgtype) > 0)
            assert(msgtype in MESSAGEMAP)
            assert(len(remainder) == 0 or not remainder.decode().isprintable())
            length: int = int.from_bytes(tmp_header.read(LENGTH_SIZE), "little")
            data = f_in.read(length)
            assert_equal(len(data), length)



            

Reported by Bandit.

              


class MessageCaptureTest(BitcoinTestFramework):
    def set_test_params(self):
        self.num_nodes = 1
        self.extra_args = [["-capturemessages"]]
        self.setup_clean_chain = True


            

Reported by Pylint.

                int option_config_;
};

TEST(DBTest, Empty) {
  do {
    ASSERT_TRUE(db_ != nullptr);
    ASSERT_EQ("NOT_FOUND", Get("foo"));
  } while (ChangeOptions());
}

            

Reported by Cppcheck.

                  int last_non_zero_offset = 0;
    for (int level = 0; level < config::kNumLevels; level++) {
      int f = NumTableFilesAtLevel(level);
      char buf[100];
      snprintf(buf, sizeof(buf), "%s%d", (level ? "," : ""), f);
      result += buf;
      if (f > 0) {
        last_non_zero_offset = result.size();
      }

            

Reported by FlawFinder.

              }

static std::string Key(int i) {
  char buf[100];
  snprintf(buf, sizeof(buf), "key%06d", i);
  return std::string(buf);
}

TEST(DBTest, MinorCompactionsHappen) {

            

Reported by FlawFinder.

                Put("A", "va");
  // Write approximately 100MB of "B" values
  for (int i = 0; i < 100000; i++) {
    char key[100];
    snprintf(key, sizeof(key), "B%010d", i);
    Put(key, value);
  }
  Put("C", "vc");
  dbfull()->TEST_CompactMemTable();

            

Reported by FlawFinder.

              
  for (int run = 0; run < 2; run++) {
    for (int i = 0; i < 1000; i++) {
      char buf[100];
      snprintf(buf, sizeof(buf), "[%d]", i * 10);
      ASSERT_OK(Put(buf, buf));
    }
    Compact("[0]", "[1000000]");
  }

            

Reported by FlawFinder.

                fprintf(stderr, "... starting thread %d\n", id);
  Random rnd(1000 + id);
  std::string value;
  char valbuf[1500];
  while (!t->state->stop.load(std::memory_order_acquire)) {
    t->state->counter[id].store(counter, std::memory_order_release);

    int key = rnd.Uniform(kNumKeys);
    char keybuf[20];

            

Reported by FlawFinder.

                  t->state->counter[id].store(counter, std::memory_order_release);

    int key = rnd.Uniform(kNumKeys);
    char keybuf[20];
    snprintf(keybuf, sizeof(keybuf), "%016d", key);

    if (rnd.OneIn(2)) {
      // Write values of the form <key, my id, counter>.
      // We add some padding for force compactions.

            

Reported by FlawFinder.

              }

std::string MakeKey(unsigned int num) {
  char buf[30];
  snprintf(buf, sizeof(buf), "%016u", num);
  return std::string(buf);
}

void BM_LogAndApply(int iters, int num_base_files) {

            

Reported by FlawFinder.

                }
  uint64_t stop_micros = env->NowMicros();
  unsigned int us = stop_micros - start_micros;
  char buf[16];
  snprintf(buf, sizeof(buf), "%d", num_base_files);
  fprintf(stderr,
          "BM_LogAndApply/%-6s   %8d iters : %9u us (%7.0f us / iter)\n", buf,
          iters, us, ((float)us) / iters);
}

            

Reported by FlawFinder.

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

class DisableWalletTest (BitcoinTestFramework):
    def set_test_params(self):
        self.setup_clean_chain = True
        self.num_nodes = 1
        self.extra_args = [["-disablewallet"]]
        self.wallet_names = []

            

Reported by Pylint.

                  def run_test (self):
        # Make sure wallet is really disabled
        assert_raises_rpc_error(-32601, 'Method not found', self.nodes[0].getwalletinfo)
        x = self.nodes[0].validateaddress('3J98t1WpEZ73CNmQviecrnyiWrnqRhWNLy')
        assert x['isvalid'] == False
        x = self.nodes[0].validateaddress('mneYUmWYsuk7kySiURxCi3AGxrAqZxLgPZ')
        assert x['isvalid'] == True

        # Checking mining to an address without a wallet. Generating to a valid address should succeed

            

Reported by Pylint.

                      # Make sure wallet is really disabled
        assert_raises_rpc_error(-32601, 'Method not found', self.nodes[0].getwalletinfo)
        x = self.nodes[0].validateaddress('3J98t1WpEZ73CNmQviecrnyiWrnqRhWNLy')
        assert x['isvalid'] == False
        x = self.nodes[0].validateaddress('mneYUmWYsuk7kySiURxCi3AGxrAqZxLgPZ')
        assert x['isvalid'] == True

        # Checking mining to an address without a wallet. Generating to a valid address should succeed
        # but generating to an invalid address will fail.

            

Reported by Bandit.

                      # Make sure wallet is really disabled
        assert_raises_rpc_error(-32601, 'Method not found', self.nodes[0].getwalletinfo)
        x = self.nodes[0].validateaddress('3J98t1WpEZ73CNmQviecrnyiWrnqRhWNLy')
        assert x['isvalid'] == False
        x = self.nodes[0].validateaddress('mneYUmWYsuk7kySiURxCi3AGxrAqZxLgPZ')
        assert x['isvalid'] == True

        # Checking mining to an address without a wallet. Generating to a valid address should succeed
        # but generating to an invalid address will fail.

            

Reported by Pylint.

                      assert_raises_rpc_error(-32601, 'Method not found', self.nodes[0].getwalletinfo)
        x = self.nodes[0].validateaddress('3J98t1WpEZ73CNmQviecrnyiWrnqRhWNLy')
        assert x['isvalid'] == False
        x = self.nodes[0].validateaddress('mneYUmWYsuk7kySiURxCi3AGxrAqZxLgPZ')
        assert x['isvalid'] == True

        # Checking mining to an address without a wallet. Generating to a valid address should succeed
        # but generating to an invalid address will fail.
        self.nodes[0].generatetoaddress(1, 'mneYUmWYsuk7kySiURxCi3AGxrAqZxLgPZ')

            

Reported by Pylint.

                      x = self.nodes[0].validateaddress('3J98t1WpEZ73CNmQviecrnyiWrnqRhWNLy')
        assert x['isvalid'] == False
        x = self.nodes[0].validateaddress('mneYUmWYsuk7kySiURxCi3AGxrAqZxLgPZ')
        assert x['isvalid'] == True

        # Checking mining to an address without a wallet. Generating to a valid address should succeed
        # but generating to an invalid address will fail.
        self.nodes[0].generatetoaddress(1, 'mneYUmWYsuk7kySiURxCi3AGxrAqZxLgPZ')
        assert_raises_rpc_error(-5, "Invalid address", self.nodes[0].generatetoaddress, 1, '3J98t1WpEZ73CNmQviecrnyiWrnqRhWNLy')

            

Reported by Pylint.

                      x = self.nodes[0].validateaddress('3J98t1WpEZ73CNmQviecrnyiWrnqRhWNLy')
        assert x['isvalid'] == False
        x = self.nodes[0].validateaddress('mneYUmWYsuk7kySiURxCi3AGxrAqZxLgPZ')
        assert x['isvalid'] == True

        # Checking mining to an address without a wallet. Generating to a valid address should succeed
        # but generating to an invalid address will fail.
        self.nodes[0].generatetoaddress(1, 'mneYUmWYsuk7kySiURxCi3AGxrAqZxLgPZ')
        assert_raises_rpc_error(-5, "Invalid address", self.nodes[0].generatetoaddress, 1, '3J98t1WpEZ73CNmQviecrnyiWrnqRhWNLy')

            

Reported by Bandit.

                      x = self.nodes[0].validateaddress('mneYUmWYsuk7kySiURxCi3AGxrAqZxLgPZ')
        assert x['isvalid'] == True

        # Checking mining to an address without a wallet. Generating to a valid address should succeed
        # but generating to an invalid address will fail.
        self.nodes[0].generatetoaddress(1, 'mneYUmWYsuk7kySiURxCi3AGxrAqZxLgPZ')
        assert_raises_rpc_error(-5, "Invalid address", self.nodes[0].generatetoaddress, 1, '3J98t1WpEZ73CNmQviecrnyiWrnqRhWNLy')

if __name__ == '__main__':

            

Reported by Pylint.

                      # Checking mining to an address without a wallet. Generating to a valid address should succeed
        # but generating to an invalid address will fail.
        self.nodes[0].generatetoaddress(1, 'mneYUmWYsuk7kySiURxCi3AGxrAqZxLgPZ')
        assert_raises_rpc_error(-5, "Invalid address", self.nodes[0].generatetoaddress, 1, '3J98t1WpEZ73CNmQviecrnyiWrnqRhWNLy')

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

            

Reported by Pylint.

                          WitnessUnknown witness_unknown{};
            witness_unknown.version = fuzzed_data_provider.ConsumeIntegralInRange(2, 16);
            std::vector<uint8_t> witness_unknown_program_1{fuzzed_data_provider.ConsumeBytes<uint8_t>(40)};
            if (witness_unknown_program_1.size() < 2) {
                witness_unknown_program_1 = {0, 0};
            }
            witness_unknown.length = witness_unknown_program_1.size();
            std::copy(witness_unknown_program_1.begin(), witness_unknown_program_1.end(), witness_unknown.program);
            tx_destination = witness_unknown;

            

Reported by Cppcheck.

                      }
        return len;
    }
    if (m_fuzzed_data_provider.ConsumeBool() && std::getenv("FUZZED_SOCKET_FAKE_LATENCY") != nullptr) {
        std::this_thread::sleep_for(std::chrono::milliseconds{2});
    }
    return random_bytes.size();
}


            

Reported by FlawFinder.

                      }
        return r;
    }
    std::memcpy(buf, random_bytes.data(), random_bytes.size());
    if (pad_to_len_bytes) {
        if (len > random_bytes.size()) {
            std::memset((char*)buf + random_bytes.size(), 0, len - random_bytes.size());
        }
        return len;

            

Reported by FlawFinder.

                  if (opt_val == nullptr) {
        return 0;
    }
    std::memcpy(opt_val,
                ConsumeFixedLengthByteVector(m_fuzzed_data_provider, *opt_len).data(),
                *opt_len);
    return 0;
}


            

Reported by FlawFinder.

                  } else if (network == Network::NET_IPV6) {
        if (fuzzed_data_provider.remaining_bytes() >= 16) {
            in6_addr v6_addr = {};
            memcpy(v6_addr.s6_addr, fuzzed_data_provider.ConsumeBytes<uint8_t>(16).data(), 16);
            net_addr = CNetAddr{v6_addr, fuzzed_data_provider.ConsumeIntegral<uint32_t>()};
        }
    } else if (network == Network::NET_INTERNAL) {
        net_addr.SetInternal(fuzzed_data_provider.ConsumeBytesAsString(32));
    } else if (network == Network::NET_ONION) {

            

Reported by FlawFinder.

                  return net_addr;
}

FILE* FuzzedFileProvider::open()
{
    SetFuzzedErrNo(m_fuzzed_data_provider);
    if (m_fuzzed_data_provider.ConsumeBool()) {
        return nullptr;
    }

            

Reported by FlawFinder.

                  if (random_bytes.empty()) {
        return 0;
    }
    std::memcpy(buf, random_bytes.data(), random_bytes.size());
    if (AdditionOverflow(fuzzed_file->m_offset, (int64_t)random_bytes.size())) {
        return fuzzed_file->m_fuzzed_data_provider.ConsumeBool() ? 0 : -1;
    }
    fuzzed_file->m_offset += random_bytes.size();
    return random_bytes.size();

            

Reported by FlawFinder.

                      });
#if defined _GNU_SOURCE && !defined __ANDROID__
    const cookie_io_functions_t io_hooks = {
        FuzzedFileProvider::read,
        FuzzedFileProvider::write,
        FuzzedFileProvider::seek,
        FuzzedFileProvider::close,
    };
    return fopencookie(this, mode.c_str(), io_hooks);

            

Reported by FlawFinder.

              #endif
}

ssize_t FuzzedFileProvider::read(void* cookie, char* buf, size_t size)
{
    FuzzedFileProvider* fuzzed_file = (FuzzedFileProvider*)cookie;
    SetFuzzedErrNo(fuzzed_file->m_fuzzed_data_provider);
    if (buf == nullptr || size == 0 || fuzzed_file->m_fuzzed_data_provider.ConsumeBool()) {
        return fuzzed_file->m_fuzzed_data_provider.ConsumeBool() ? 0 : -1;

            

Reported by FlawFinder.

              
    void Finalize(Span<unsigned char> output) {
        assert(output.size() == OUTPUT_SIZE);
        unsigned char buf[CSHA256::OUTPUT_SIZE];
        sha.Finalize(buf);
        sha.Reset().Write(buf, CSHA256::OUTPUT_SIZE).Finalize(output.data());
    }

    CHash256& Write(Span<const unsigned char> input) {

            

Reported by FlawFinder.

              
    void Finalize(Span<unsigned char> output) {
        assert(output.size() == OUTPUT_SIZE);
        unsigned char buf[CSHA256::OUTPUT_SIZE];
        sha.Finalize(buf);
        CRIPEMD160().Write(buf, CSHA256::OUTPUT_SIZE).Finalize(output.data());
    }

    CHash160& Write(Span<const unsigned char> input) {

            

Reported by FlawFinder.

              
    void ignore(size_t nSize)
    {
        char data[1024];
        while (nSize > 0) {
            size_t now = std::min<size_t>(nSize, 1024);
            read(data, now);
            nSize -= now;
        }

            

Reported by FlawFinder.

              
unsigned int MurmurHash3(unsigned int nHashSeed, Span<const unsigned char> vDataToHash);

void BIP32Hash(const ChainCode &chainCode, unsigned int nChild, unsigned char header, const unsigned char data[32], unsigned char output[64]);

/** Return a CHashWriter primed for tagged hashes (as specified in BIP 340).
 *
 * The returned object will have SHA256(tag) written to it twice (= 64 bytes).
 * A tagged hash can be computed by feeding the message into this object, and

            

Reported by FlawFinder.

              
unsigned int MurmurHash3(unsigned int nHashSeed, Span<const unsigned char> vDataToHash);

void BIP32Hash(const ChainCode &chainCode, unsigned int nChild, unsigned char header, const unsigned char data[32], unsigned char output[64]);

/** Return a CHashWriter primed for tagged hashes (as specified in BIP 340).
 *
 * The returned object will have SHA256(tag) written to it twice (= 64 bytes).
 * A tagged hash can be computed by feeding the message into this object, and

            

Reported by FlawFinder.

              
unsigned int MurmurHash3(unsigned int nHashSeed, Span<const unsigned char> vDataToHash);

void BIP32Hash(const ChainCode &chainCode, unsigned int nChild, unsigned char header, const unsigned char data[32], unsigned char output[64]);

/** Return a CHashWriter primed for tagged hashes (as specified in BIP 340).
 *
 * The returned object will have SHA256(tag) written to it twice (= 64 bytes).
 * A tagged hash can be computed by feeding the message into this object, and

            

Reported by FlawFinder.

              public:
    explicit CHashVerifier(Source* source_) : CHashWriter(source_->GetType(), source_->GetVersion()), source(source_) {}

    void read(char* pch, size_t nSize)
    {
        source->read(pch, nSize);
        this->write(pch, nSize);
    }


            

Reported by FlawFinder.

              
    void read(char* pch, size_t nSize)
    {
        source->read(pch, nSize);
        this->write(pch, nSize);
    }

    void ignore(size_t nSize)
    {

            

Reported by FlawFinder.

                      char data[1024];
        while (nSize > 0) {
            size_t now = std::min<size_t>(nSize, 1024);
            read(data, now);
            nSize -= now;
        }
    }

    template<typename T>

            

Reported by FlawFinder.

              
namespace fsbridge {

FILE *fopen(const fs::path& p, const char *mode)
{
#ifndef WIN32
    return ::fopen(p.string().c_str(), mode);
#else
    std::wstring_convert<std::codecvt_utf8_utf16<wchar_t>,wchar_t> utf8_cvt;

            

Reported by FlawFinder.

              FILE *fopen(const fs::path& p, const char *mode)
{
#ifndef WIN32
    return ::fopen(p.string().c_str(), mode);
#else
    std::wstring_convert<std::codecvt_utf8_utf16<wchar_t>,wchar_t> utf8_cvt;
    return ::_wfopen(p.wstring().c_str(), utf8_cvt.from_bytes(mode).c_str());
#endif
}

            

Reported by FlawFinder.

              
FileLock::FileLock(const fs::path& file)
{
    fd = open(file.string().c_str(), O_RDWR);
    if (fd == -1) {
        reason = GetErrorReason();
    }
}


            

Reported by FlawFinder.

              #else
    // Convert from Multi Byte to utf-16
    std::string mb_string(e.what());
    int size = MultiByteToWideChar(CP_ACP, 0, mb_string.data(), mb_string.size(), nullptr, 0);

    std::wstring utf16_string(size, L'\0');
    MultiByteToWideChar(CP_ACP, 0, mb_string.data(), mb_string.size(), &*utf16_string.begin(), size);
    // Convert from utf-16 to utf-8
    return std::wstring_convert<std::codecvt_utf8_utf16<wchar_t>, wchar_t>().to_bytes(utf16_string);

            

Reported by FlawFinder.

                  int size = MultiByteToWideChar(CP_ACP, 0, mb_string.data(), mb_string.size(), nullptr, 0);

    std::wstring utf16_string(size, L'\0');
    MultiByteToWideChar(CP_ACP, 0, mb_string.data(), mb_string.size(), &*utf16_string.begin(), size);
    // Convert from utf-16 to utf-8
    return std::wstring_convert<std::codecvt_utf8_utf16<wchar_t>, wchar_t>().to_bytes(utf16_string);
#endif
}


            

Reported by FlawFinder.

                  }
}

void ifstream::open(const fs::path& p, std::ios_base::openmode mode)
{
    close();
    mode |= std::ios_base::in;
    m_file = fsbridge::fopen(p, openmodeToStr(mode).c_str());
    if (m_file == nullptr) {

            

Reported by FlawFinder.

              {
    close();
    mode |= std::ios_base::in;
    m_file = fsbridge::fopen(p, openmodeToStr(mode).c_str());
    if (m_file == nullptr) {
        return;
    }
    m_filebuf = __gnu_cxx::stdio_filebuf<char>(m_file, mode);
    rdbuf(&m_filebuf);

            

Reported by FlawFinder.

                  m_file = nullptr;
}

void ofstream::open(const fs::path& p, std::ios_base::openmode mode)
{
    close();
    mode |= std::ios_base::out;
    m_file = fsbridge::fopen(p, openmodeToStr(mode).c_str());
    if (m_file == nullptr) {

            

Reported by FlawFinder.

              {
    close();
    mode |= std::ios_base::out;
    m_file = fsbridge::fopen(p, openmodeToStr(mode).c_str());
    if (m_file == nullptr) {
        return;
    }
    m_filebuf = __gnu_cxx::stdio_filebuf<char>(m_file, mode);
    rdbuf(&m_filebuf);

            

Reported by FlawFinder.

              MAX_FUTURE_BLOCK_TIME = 2 * 60 * 60


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


            

Reported by Pylint.

                      self.setup_clean_chain = True
        self.extra_args = [["-whitelist=noban@127.0.0.1"]]

    def run_test(self):
        # Add p2p connection to node0
        node = self.nodes[0]  # convenience reference to the node
        peer = node.add_p2p_connection(P2PDataStore())

        best_block = node.getblock(node.getbestblockhash())

            

Reported by Pylint.

                      self.setup_clean_chain = True
        self.extra_args = [["-whitelist=noban@127.0.0.1"]]

    def run_test(self):
        # Add p2p connection to node0
        node = self.nodes[0]  # convenience reference to the node
        peer = node.add_p2p_connection(P2PDataStore())

        best_block = node.getblock(node.getbestblockhash())

            

Reported by Pylint.

                      block2.vtx.append(tx2)
        assert_equal(block2.hashMerkleRoot, block2.calc_merkle_root())
        assert_equal(orig_hash, block2.rehash())
        assert block2_orig.vtx != block2.vtx

        peer.send_blocks_and_test([block2], node, success=False, reject_reason='bad-txns-duplicate')

        # Check transactions for duplicate inputs (CVE-2018-17144)
        self.log.info("Test duplicate input block.")

            

Reported by Bandit.

                      block2_dup.hashMerkleRoot = block2_dup.calc_merkle_root()
        block2_dup.rehash()
        block2_dup.solve()
        peer.send_blocks_and_test([block2_dup], node, success=False, reject_reason='bad-txns-inputs-duplicate')

        self.log.info("Test very broken block.")

        block3 = create_block(tip, create_coinbase(height), block_time)
        block_time += 1

            

Reported by Pylint.

                      block4.rehash()
        block4.solve()
        self.log.info("Test inflation by duplicating input")
        peer.send_blocks_and_test([block4], node, success=False,  reject_reason='bad-txns-inputs-duplicate')

        self.log.info("Test accepting identical block after rejecting it due to a future timestamp.")
        t = int(time.time())
        node.setmocktime(t)
        # Set block time +1 second past max future validity

            

Reported by Pylint.

                      self.log.info("Test inflation by duplicating input")
        peer.send_blocks_and_test([block4], node, success=False,  reject_reason='bad-txns-inputs-duplicate')

        self.log.info("Test accepting identical block after rejecting it due to a future timestamp.")
        t = int(time.time())
        node.setmocktime(t)
        # Set block time +1 second past max future validity
        block = create_block(tip, create_coinbase(height), t + MAX_FUTURE_BLOCK_TIME + 1)
        block.hashMerkleRoot = block.calc_merkle_root()

            

Reported by Pylint.

                      peer.send_blocks_and_test([block4], node, success=False,  reject_reason='bad-txns-inputs-duplicate')

        self.log.info("Test accepting identical block after rejecting it due to a future timestamp.")
        t = int(time.time())
        node.setmocktime(t)
        # Set block time +1 second past max future validity
        block = create_block(tip, create_coinbase(height), t + MAX_FUTURE_BLOCK_TIME + 1)
        block.hashMerkleRoot = block.calc_merkle_root()
        block.solve()

            

Reported by Pylint.

                      block.hashMerkleRoot = block.calc_merkle_root()
        block.solve()
        # Need force_send because the block will get rejected without a getdata otherwise
        peer.send_blocks_and_test([block], node, force_send=True, success=False, reject_reason='time-too-new')
        node.setmocktime(t + 1)
        peer.send_blocks_and_test([block], node, success=True)


if __name__ == '__main__':

            

Reported by Pylint.

              
    def run_test(self):
        datadir = os.path.join(self.nodes[0].datadir, self.chain)
        self.log.info("Using datadir {}".format(datadir))

        self.log.info("Check that we can't start a second bitcoind instance using the same datadir")
        expected_msg = "Error: Cannot obtain a lock on data directory {0}. {1} is probably already running.".format(datadir, self.config['environment']['PACKAGE_NAME'])
        self.nodes[1].assert_start_raises_init_error(extra_args=['-datadir={}'.format(self.nodes[0].datadir), '-noserver'], expected_msg=expected_msg)


            

Reported by Pylint.

              # 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.
"""Check that it's not possible to start a second bitcoind instance using the same datadir or wallet."""
import os
import random
import string

from test_framework.test_framework import BitcoinTestFramework

            

Reported by Pylint.

              from test_framework.test_framework import BitcoinTestFramework
from test_framework.test_node import ErrorMatch

class FilelockTest(BitcoinTestFramework):
    def set_test_params(self):
        self.setup_clean_chain = True
        self.num_nodes = 2

    def setup_network(self):

            

Reported by Pylint.

                      self.log.info("Using datadir {}".format(datadir))

        self.log.info("Check that we can't start a second bitcoind instance using the same datadir")
        expected_msg = "Error: Cannot obtain a lock on data directory {0}. {1} is probably already running.".format(datadir, self.config['environment']['PACKAGE_NAME'])
        self.nodes[1].assert_start_raises_init_error(extra_args=['-datadir={}'.format(self.nodes[0].datadir), '-noserver'], expected_msg=expected_msg)

        if self.is_wallet_compiled():
            def check_wallet_filelock(descriptors):
                wallet_name = ''.join([random.choice(string.ascii_lowercase) for _ in range(6)])

            

Reported by Pylint.

              
        self.log.info("Check that we can't start a second bitcoind instance using the same datadir")
        expected_msg = "Error: Cannot obtain a lock on data directory {0}. {1} is probably already running.".format(datadir, self.config['environment']['PACKAGE_NAME'])
        self.nodes[1].assert_start_raises_init_error(extra_args=['-datadir={}'.format(self.nodes[0].datadir), '-noserver'], expected_msg=expected_msg)

        if self.is_wallet_compiled():
            def check_wallet_filelock(descriptors):
                wallet_name = ''.join([random.choice(string.ascii_lowercase) for _ in range(6)])
                self.nodes[0].createwallet(wallet_name=wallet_name, descriptors=descriptors)

            

Reported by Pylint.

              
        if self.is_wallet_compiled():
            def check_wallet_filelock(descriptors):
                wallet_name = ''.join([random.choice(string.ascii_lowercase) for _ in range(6)])
                self.nodes[0].createwallet(wallet_name=wallet_name, descriptors=descriptors)
                wallet_dir = os.path.join(datadir, 'wallets')
                self.log.info("Check that we can't start a second bitcoind instance using the same wallet")
                if descriptors:
                    expected_msg = "Error: SQLiteDatabase: Unable to obtain an exclusive lock on the database, is it being used by another bitcoind?"

            

Reported by Bandit.

                              wallet_name = ''.join([random.choice(string.ascii_lowercase) for _ in range(6)])
                self.nodes[0].createwallet(wallet_name=wallet_name, descriptors=descriptors)
                wallet_dir = os.path.join(datadir, 'wallets')
                self.log.info("Check that we can't start a second bitcoind instance using the same wallet")
                if descriptors:
                    expected_msg = "Error: SQLiteDatabase: Unable to obtain an exclusive lock on the database, is it being used by another bitcoind?"
                else:
                    expected_msg = "Error: Error initializing wallet database environment"
                self.nodes[1].assert_start_raises_init_error(extra_args=['-walletdir={}'.format(wallet_dir), '-wallet=' + wallet_name, '-noserver'], expected_msg=expected_msg, match=ErrorMatch.PARTIAL_REGEX)

            

Reported by Pylint.

                              wallet_dir = os.path.join(datadir, 'wallets')
                self.log.info("Check that we can't start a second bitcoind instance using the same wallet")
                if descriptors:
                    expected_msg = "Error: SQLiteDatabase: Unable to obtain an exclusive lock on the database, is it being used by another bitcoind?"
                else:
                    expected_msg = "Error: Error initializing wallet database environment"
                self.nodes[1].assert_start_raises_init_error(extra_args=['-walletdir={}'.format(wallet_dir), '-wallet=' + wallet_name, '-noserver'], expected_msg=expected_msg, match=ErrorMatch.PARTIAL_REGEX)

            if self.is_bdb_compiled():

            

Reported by Pylint.

                                  expected_msg = "Error: SQLiteDatabase: Unable to obtain an exclusive lock on the database, is it being used by another bitcoind?"
                else:
                    expected_msg = "Error: Error initializing wallet database environment"
                self.nodes[1].assert_start_raises_init_error(extra_args=['-walletdir={}'.format(wallet_dir), '-wallet=' + wallet_name, '-noserver'], expected_msg=expected_msg, match=ErrorMatch.PARTIAL_REGEX)

            if self.is_bdb_compiled():
                check_wallet_filelock(False)
            if self.is_sqlite_compiled():
                check_wallet_filelock(True)

            

Reported by Pylint.

                }

  Status GetTestDirectory(std::string* result) override {
    const char* env = getenv("TEST_TMPDIR");
    if (env && env[0] != '\0') {
      *result = env;
      return Status::OK();
    }


            

Reported by FlawFinder.

              
    // Fit as much as possible into buffer.
    size_t copy_size = std::min(write_size, kWritableFileBufferSize - pos_);
    std::memcpy(buf_ + pos_, write_data, copy_size);
    write_data += copy_size;
    write_size -= copy_size;
    pos_ += copy_size;
    if (write_size == 0) {
      return Status::OK();

            

Reported by FlawFinder.

              
    // Small writes go to buffer, large writes are written directly.
    if (write_size < kWritableFileBufferSize) {
      std::memcpy(buf_, write_data, write_size);
      pos_ = write_size;
      return Status::OK();
    }
    return WriteUnbuffered(write_data, write_size);
  }

            

Reported by FlawFinder.

                }

  // buf_[0, pos_-1] contains data to be written to handle_.
  char buf_[kWritableFileBufferSize];
  size_t pos_;

  ScopedHandle handle_;
  const std::string filename_;
};

            

Reported by FlawFinder.

                    return WindowsError(directory_path, last_error);
    }
    do {
      char base_name[_MAX_FNAME];
      char ext[_MAX_EXT];

      auto find_data_filename = toUtf8(find_data.cFileName);
      if (!_splitpath_s(find_data_filename.c_str(), nullptr, 0, nullptr, 0,
                        base_name, ARRAYSIZE(base_name), ext, ARRAYSIZE(ext))) {

            

Reported by FlawFinder.

                  }
    do {
      char base_name[_MAX_FNAME];
      char ext[_MAX_EXT];

      auto find_data_filename = toUtf8(find_data.cFileName);
      if (!_splitpath_s(find_data_filename.c_str(), nullptr, 0, nullptr, 0,
                        base_name, ARRAYSIZE(base_name), ext, ARRAYSIZE(ext))) {
        result->emplace_back(std::string(base_name) + ext);

            

Reported by FlawFinder.

                    return Status::OK();
    }

    wchar_t wtmp_path[MAX_PATH];
    if (!GetTempPathW(ARRAYSIZE(wtmp_path), wtmp_path)) {
      return WindowsError("GetTempPath", ::GetLastError());
    }
    std::string tmp_path = toUtf8(std::wstring(wtmp_path));
    std::stringstream ss;

            

Reported by FlawFinder.

                std::wstring toUtf16(const std::string& str) {
    if (str.empty()) return std::wstring();
    int size_needed =
        MultiByteToWideChar(CP_UTF8, 0, &str[0], (int)str.size(), NULL, 0);
    std::wstring strTo(size_needed, 0);
    MultiByteToWideChar(CP_UTF8, 0, &str[0], (int)str.size(), &strTo[0],
                        size_needed);
    return strTo;
  }

            

Reported by FlawFinder.

                  int size_needed =
        MultiByteToWideChar(CP_UTF8, 0, &str[0], (int)str.size(), NULL, 0);
    std::wstring strTo(size_needed, 0);
    MultiByteToWideChar(CP_UTF8, 0, &str[0], (int)str.size(), &strTo[0],
                        size_needed);
    return strTo;
  }
};


            

Reported by FlawFinder.

              
/* algo argument for nonce_function_bip340 to derive the nonce exactly as stated in BIP-340
 * by using the correct tagged hash function. */
static const unsigned char bip340_algo[13] = "BIP0340/nonce";

static const unsigned char schnorrsig_extraparams_magic[4] = SECP256K1_SCHNORRSIG_EXTRAPARAMS_MAGIC;

static int nonce_function_bip340(unsigned char *nonce32, const unsigned char *msg, size_t msglen, const unsigned char *key32, const unsigned char *xonly_pk32, const unsigned char *algo, size_t algolen, void *data) {
    secp256k1_sha256 sha;

            

Reported by FlawFinder.

               * by using the correct tagged hash function. */
static const unsigned char bip340_algo[13] = "BIP0340/nonce";

static const unsigned char schnorrsig_extraparams_magic[4] = SECP256K1_SCHNORRSIG_EXTRAPARAMS_MAGIC;

static int nonce_function_bip340(unsigned char *nonce32, const unsigned char *msg, size_t msglen, const unsigned char *key32, const unsigned char *xonly_pk32, const unsigned char *algo, size_t algolen, void *data) {
    secp256k1_sha256 sha;
    unsigned char masked_key[32];
    int i;

            

Reported by FlawFinder.

              
static int nonce_function_bip340(unsigned char *nonce32, const unsigned char *msg, size_t msglen, const unsigned char *key32, const unsigned char *xonly_pk32, const unsigned char *algo, size_t algolen, void *data) {
    secp256k1_sha256 sha;
    unsigned char masked_key[32];
    int i;

    if (algo == NULL) {
        return 0;
    }

            

Reported by FlawFinder.

              
static void secp256k1_schnorrsig_challenge(secp256k1_scalar* e, const unsigned char *r32, const unsigned char *msg, size_t msglen, const unsigned char *pubkey32)
{
    unsigned char buf[32];
    secp256k1_sha256 sha;

    /* tagged hash(r.x, pk.x, msg) */
    secp256k1_schnorrsig_sha256_tagged(&sha);
    secp256k1_sha256_write(&sha, r32, 32);

            

Reported by FlawFinder.

                  secp256k1_gej rj;
    secp256k1_ge pk;
    secp256k1_ge r;
    unsigned char buf[32] = { 0 };
    unsigned char pk_buf[32];
    unsigned char seckey[32];
    int ret = 1;

    VERIFY_CHECK(ctx != NULL);

            

Reported by FlawFinder.

                  secp256k1_ge pk;
    secp256k1_ge r;
    unsigned char buf[32] = { 0 };
    unsigned char pk_buf[32];
    unsigned char seckey[32];
    int ret = 1;

    VERIFY_CHECK(ctx != NULL);
    ARG_CHECK(secp256k1_ecmult_gen_context_is_built(&ctx->ecmult_gen_ctx));

            

Reported by FlawFinder.

                  secp256k1_ge r;
    unsigned char buf[32] = { 0 };
    unsigned char pk_buf[32];
    unsigned char seckey[32];
    int ret = 1;

    VERIFY_CHECK(ctx != NULL);
    ARG_CHECK(secp256k1_ecmult_gen_context_is_built(&ctx->ecmult_gen_ctx));
    ARG_CHECK(sig64 != NULL);

            

Reported by FlawFinder.

                  secp256k1_gej pkj;
    secp256k1_fe rx;
    secp256k1_ge r;
    unsigned char buf[32];
    int overflow;

    VERIFY_CHECK(ctx != NULL);
    ARG_CHECK(secp256k1_ecmult_context_is_built(&ctx->ecmult_ctx));
    ARG_CHECK(sig64 != NULL);

            

Reported by FlawFinder.