thehub/libs/p2p/PeerAddressDB.cpp

/*
 * This file is part of the Flowee project
 * Copyright (C) 2020-2024 Tom Zander <tom@flowee.org>
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
#include "PeerAddressDB.h"
#include "ConnectionManager.h"
#include "Peer.h"
#include <Message.h>
#include <streaming/BufferPool.h>
#include <streaming/BufferPools.h>
#include <streaming/MessageBuilder.h>
#include <streaming/MessageParser.h>
#include <streaming/P2PParser.h>
#include <random.h>

#include <fstream>

enum SavingTags {
    Separator = 0,
    Hostname,
    IPAddress,
    Port,
    Services,
    LastConnected,
    Punishment,
    Segment,
    EverConnected,
    LastReceivedGoodHeaders,
    // AskedAddr?

    DBVersion = 40,
};

const EndPoint &PeerAddress::peerAddress() const
{
    auto i = d->m_peers.find(m_id);
    assert(d->m_peers.end() != i);
    return i->second.address;
}

void PeerAddress::successfullyConnected(uint64_t services)
{
    auto i = d->m_peers.find(m_id);
    assert(d->m_peers.end() != i);
    time_t now = time(nullptr);
    assert(now > 0);
    i->second.lastConnected = static_cast<uint32_t>(now);
    i->second.services = services;
    if (i->second.punishment >= 200)
        i->second.punishment -= 125;
    i->second.inUse = true;
    i->second.everConnected = true;
    d->m_needsSave = true;
}

void PeerAddress::gotGoodHeaders()
{
    auto i = d->m_peers.find(m_id);
    assert(d->m_peers.end() != i);
    time_t now = time(nullptr);
    assert(now > 0);
    i->second.lastConnected = static_cast<uint32_t>(now);
    if (i->second.punishment > 300)
        i->second.punishment -= 270;
    i->second.lastReceivedGoodHeaders = now;
    d->m_needsSave = true;
}

short PeerAddress::punishPeer(short amount)
{
    auto i = d->m_peers.find(m_id);
    assert(d->m_peers.end() != i);
    int newPunishment = int(i->second.punishment) + amount;
    if (i->second.punishment < PUNISHMENT_MAX && newPunishment >= PUNISHMENT_MAX) {
        d->m_disabledPeerCount++;
    } else if (i->second.punishment >= PUNISHMENT_MAX && newPunishment < PUNISHMENT_MAX) {
        d->m_disabledPeerCount--;
    }

    i->second.punishment = std::min(newPunishment, 0xeFFF); // avoid overflow
    d->m_needsSave = true;
    return i->second.punishment;
}

short PeerAddress::punishment() const
{
    auto i = d->m_peers.find(m_id);
    assert(d->m_peers.end() != i);
    return i->second.punishment;
}

void PeerAddress::resetPunishment()
{
    auto i = d->m_peers.find(m_id);
    assert(d->m_peers.end() != i);
    i->second.punishment = 0;
    // update lastConnected to now without setting 'everConnected' to
    // at least remember when we reset the punishment.
    i->second.lastConnected = static_cast<uint32_t>(time(nullptr));
    d->m_needsSave = true;
}

bool PeerAddress::isValid() const
{
    return d && m_id >= 0 && d->m_nextPeerId > m_id;
}

bool PeerAddress::askedAddresses() const
{
    auto i = d->m_peers.find(m_id);
    assert(d->m_peers.end() != i);
    return i->second.askedAddr;
}

void PeerAddress::setAskedAddresses(bool on)
{
    auto i = d->m_peers.find(m_id);
    assert(d->m_peers.end() != i);
    i->second.askedAddr = on;
    d->m_needsSave = true;
}

bool PeerAddress::hasEverConnected() const
{
    auto i = d->m_peers.find(m_id);
    assert(d->m_peers.end() != i);
    return i->second.everConnected;
}

uint32_t PeerAddress::lastReceivedGoodHeaders() const
{
    auto i = d->m_peers.find(m_id);
    assert(d->m_peers.end() != i);
    return i->second.lastReceivedGoodHeaders;
}

uint16_t PeerAddress::segment() const
{
    auto i = d->m_peers.find(m_id);
    assert(d->m_peers.end() != i);
    return i->second.segment;
}

void PeerAddress::setSegment(uint16_t segment)
{
    auto i = d->m_peers.find(m_id);
    assert(d->m_peers.end() != i);
    i->second.segment = segment;
    d->m_needsSave = true;
}

void PeerAddress::setInUse(bool on)
{
    // this is an in-memory-only property
    auto i = d->m_peers.find(m_id);
    assert(d->m_peers.end() != i);
    i->second.inUse = on;
}

void PeerAddress::setServices(uint64_t services)
{
    auto i = d->m_peers.find(m_id);
    assert(d->m_peers.end() != i);
    i->second.services = services;
    d->m_needsSave = true;
}

uint32_t PeerAddress::lastConnected() const
{
    auto i = d->m_peers.find(m_id);
    assert(d->m_peers.end() != i);
    return i->second.lastConnected;
}

PeerAddress::PeerAddress(PeerAddressDB *parent, int peerId)
    : d(parent), m_id(peerId)
{
    assert(parent);
}

int PeerAddress::id() const
{
    return m_id;
}


// ////////////////////////////////////////

PeerAddressDB::PeerAddressDB(ConnectionManager *parent)
    : m_parent(parent)
{
}

PeerAddress PeerAddressDB::findBest(uint64_t requiredServices, uint16_t segment)
{
    std::unique_lock<std::mutex> lock(m_lock);
    if (m_nextPeerId == 0)
        return PeerAddress(this, -1);

    int usefulCount = 0;
    int best = -1;
    int bestScore = 0;
    const uint64_t now = time(nullptr);
    for (int attempts = 0; attempts < 10000 && usefulCount < 1000; ++attempts) {
        const int i = GetRandInt(m_nextPeerId - 1);
        assert(i >= 0);
        assert(m_peers.size() > static_cast<size_t>(i));
        const PeerInfo &info = m_peers[i];
        if (info.inUse)
            continue;
        if (info.everConnected && (info.services & requiredServices) != requiredServices)
            continue;
        if (segment != 0 && info.segment != 0 && segment != info.segment)
            continue;
        if (!m_supportIPv4Net && info.address.ipAddress.is_v4())
            continue;
        if (!m_supportIPv6Net && info.address.ipAddress.is_v6())
            continue;

        ++usefulCount;
        const uint64_t hoursAgoConnected = (now - info.lastConnected) / 3600;
        /*
         * A punishment-score (aka ban-score) isn't everything.
         * If its been a long time since we connected, maybe the peer
         * has changed to be better.
         */
        int punishment = info.punishment;
        if (info.everConnected && hoursAgoConnected > 7 * 24 * 6)
            punishment /= 2;
        int score = std::max(0, 1000 - punishment);

        // If its all the same, I prefer nodes that we recently connected to (3 weeks)
        if (info.everConnected)
            score += static_cast<int32_t>(250 - std::min<uint64_t>(250, hoursAgoConnected / 2));

        if (info.address.announcePort != m_defaultPortNr) // SLIGHTLY prefer non-default port
            score += 50;

        if (score > bestScore) {
            bestScore = score;
            best = i;
        }
    }
    return PeerAddress(this, best);
}

void PeerAddressDB::resetAllStats()
{
    std::unique_lock<std::mutex> lock(m_lock);
    m_disabledPeerCount = 0;
    for (auto iter = m_peers.begin(); iter != m_peers.end(); ++iter) {
        iter->second.punishment = 0;
        iter->second.everConnected = false;
        iter->second.lastReceivedGoodHeaders = 0;
        iter->second.segment = 0;
        iter->second.askedAddr = false;
        m_needsSave = true;
    }
}

int PeerAddressDB::peerCount() const
{
    std::unique_lock<std::mutex> lock(m_lock);
    return static_cast<int>(m_peers.size()) - m_disabledPeerCount;
}

void PeerAddressDB::processAddressMessage(const Message &message, int sourcePeerId)
{
    auto sourcePeer = m_parent->peer(sourcePeerId);
    if (sourcePeer.get() == nullptr) // since disconnected.
        return;
    // make sure that this peer is actually following our chain.
    const uint32_t goodHeaders = sourcePeer->peerAddress().lastReceivedGoodHeaders();
    if (goodHeaders == 0) // ignore addresses from unvalidated peer
        return;
     // if we haven't validated them for 60 days, ignore message.
    if (time(nullptr) - goodHeaders > 3600 * 24 * 60)
        return;

    std::unique_lock<std::mutex> lock(m_lock);
    size_t oldCount = m_peers.size();
    try {
        Streaming::P2PParser parser(message);
        const size_t count = parser.readCompactInt();
        logInfo() << "Received" << count << "addresses" << "from peer:" << sourcePeerId;
        for (size_t i = 0; i < count; ++i) {
            PeerInfo info;
            info.lastConnected = parser.readInt();
            info.services = parser.readLong();
            auto ip = parser.readBytes(16);
            auto port = parser.readWordBE();
            info.address = EndPoint::fromAddr(ip, port);
            insert(info);
        }
    } catch (const std::runtime_error &err) {
        logInfo() << "Failed to read address message from peer:" << sourcePeerId;
        m_parent->punish(sourcePeerId, 250);
        return;
    }
    if (oldCount != m_peers.size()) {
        logInfo().nospace() << "We now have " << m_peers.size() << " addresses (thanks! peer: " << sourcePeerId << ")";
        m_needsSave = true;
    }
}

void PeerAddressDB::addOne(const EndPoint &endPoint)
{
    std::unique_lock<std::mutex> lock(m_lock);
    PeerInfo info;
    info.address = endPoint;
    info.services = 5;
    insert(info);
    m_needsSave = true;
}

void PeerAddressDB::saveDatabase(const boost::filesystem::path &basedir)
{
    std::unique_lock<std::mutex> lock(m_lock);
    if (!m_needsSave)
        return;
    Streaming::MessageBuilder builder(Streaming::pool(m_peers.size() * 40));
    builder.add(DBVersion, 1);
    char ip[16];
    for (const auto &item : m_peers) {
        if (item.second.address.ipAddress.is_unspecified()) {
            builder.add(Hostname, item.second.address.hostname);
        } else {
            item.second.address.toAddr(ip);
            builder.addByteArray(IPAddress, ip, 16);
        }
        if (item.second.address.announcePort != m_defaultPortNr)
            builder.add(Port, item.second.address.announcePort);
        builder.add(Services, item.second.services);
        builder.add(LastConnected, uint64_t(item.second.lastConnected));
        if (item.second.punishment > 0)
            builder.add(Punishment, item.second.punishment);
        if (item.second.segment != 0)
            builder.add(Segment, item.second.segment);
        if (item.second.everConnected)
            builder.add(EverConnected, true);
        if (item.second.lastReceivedGoodHeaders)
            builder.add(LastReceivedGoodHeaders, uint64_t(item.second.lastReceivedGoodHeaders));

        builder.add(Separator, true); // separator
    }
    auto data = builder.buffer();

    try {
        boost::system::error_code error;
        boost::filesystem::create_directories(basedir, error);
        if (error && !boost::filesystem::exists(basedir) && !boost::filesystem::is_directory(basedir)) {
            logFatal() << "P2P.PeerAddressDB can't save. Failed creating the dir:" << basedir.string();
            return;
        }
        boost::filesystem::remove(basedir / "peers.dat~");

        std::ofstream outFile((basedir / "peers.dat~").string());
        outFile.write(data.begin(), data.size());

        boost::filesystem::rename(basedir / "peers.dat~", basedir / "peers.dat");
        m_needsSave = false;
    } catch (const std::exception &e) {
        logFatal() << "Failed to save the database. Reason:" << e.what();
    }
}

void PeerAddressDB::loadDatabase(const boost::filesystem::path &basedir)
{
    std::unique_lock<std::mutex> lock(m_lock);
    std::ifstream in((basedir / "peers.dat").string());
    if (!in.is_open())
        return;
    m_nextPeerId = 0;
    const auto dataSize = boost::filesystem::file_size(basedir / "peers.dat");
    auto pool = Streaming::pool(dataSize);
    in.read(pool->begin(), dataSize);
    Streaming::MessageParser parser(pool->commit(dataSize));
    PeerInfo info;
    int version = 0;
    while (parser.next() == Streaming::FoundTag) {
        if (parser.tag() == Separator) {
            if (info.address.isValid())
                m_peers.insert(std::make_pair(m_nextPeerId++, info));
            info = PeerInfo();
        }
        else if (parser.tag() == DBVersion) {
            version = parser.intData();
        }
        else if (parser.tag() == IPAddress) {
            info.address = EndPoint::fromAddr(parser.bytesData(), m_defaultPortNr);
        }
        else if (parser.tag() == Hostname) {
            info.address = EndPoint(parser.stringData(), m_defaultPortNr);
        }
        else if (parser.tag() == Port) {
            info.address.announcePort = info.address.peerPort = parser.intData();
        }
        else if (parser.tag() == Services) {
            info.services = parser.longData();
        }
        else if (parser.tag() == LastConnected) {
            info.lastConnected = parser.longData();
        }
        else if (parser.tag() == Punishment) {
            info.punishment = parser.intData();
            if (info.punishment >= PUNISHMENT_MAX)
                m_disabledPeerCount++;
        }
        else if (parser.tag() == Segment) {
            info.segment = parser.intData();
        }
        else if (parser.tag() == EverConnected) {
            info.everConnected = parser.boolData();
        }
        else if (parser.tag() == LastReceivedGoodHeaders) {
            info.lastReceivedGoodHeaders = parser.longData();
        }
    }
    m_needsSave = false;

    if (version == 0) {
        // the old version had bugs, we will reset all the 'everConnected' to false
        // and restart fresh with the punishment too.
        for (auto iter = m_peers.begin(); iter != m_peers.end(); ++iter) {
            iter->second.punishment = 0;
            iter->second.everConnected = false;
        }
        m_disabledPeerCount = 0;
        m_needsSave = true;
    }

    logInfo() << "Peer database loaded:" << m_peers.size() << "disabled peer count:" << m_disabledPeerCount;
}

void PeerAddressDB::insert(PeerInfo &pi)
{
    if (pi.address.ipAddress.is_unspecified()) // try to see if hostname is an IP. If so, bypass DNS lookup
        try { pi.address.ipAddress = boost::asio::ip::address::from_string(pi.address.hostname); } catch (...) {}
    const bool hasIp = !pi.address.ipAddress.is_unspecified();
    if (!hasIp && pi.address.hostname.empty())
        return;
    for (auto i = m_peers.begin(); i != m_peers.end(); ++i) {
        if (hasIp && i->second.address.ipAddress == pi.address.ipAddress) {
            if (i->second.punishment > 400)
                i->second.punishment -= 100;
            return;
        }
        if (!hasIp && i->second.address.hostname == pi.address.hostname)
            return;
    }

    m_peers.insert(std::make_pair(m_nextPeerId++, pi));
    m_needsSave = true;
}

bool PeerAddressDB::supportIPv6Net() const
{
    return m_supportIPv6Net;
}

void PeerAddressDB::setSupportIPv6Net(bool on)
{
    m_supportIPv6Net = on;
}

bool PeerAddressDB::supportIPv4Net() const
{
    return m_supportIPv4Net;
}

void PeerAddressDB::setSupportIPv4Net(bool on)
{
    m_supportIPv4Net = on;
}

int PeerAddressDB::defaultPortNr() const
{
    return m_defaultPortNr;
}

void PeerAddressDB::setDefaultPortNr(int defaultPortNr)
{
    m_defaultPortNr = defaultPortNr;
}

AddressDBStats PeerAddressDB::createStats() const
{
    std::unique_lock<std::mutex> lock(m_lock);
    AddressDBStats stats;
    stats.count = m_peers.size();
    stats.banned = 0;
    stats.partialBanned = 0;
    stats.ipv6Addresses = 0;
    stats.everConnected = 0;
    stats.usesIPv4 = m_supportIPv4Net;
    stats.usesIPv6 = m_supportIPv6Net;
    for (auto iter = m_peers.begin(); iter != m_peers.end(); ++iter) {
        if (iter->second.everConnected)
            stats.everConnected++;
        const auto punishment = iter->second.punishment;
        static_assert(PUNISHMENT_MAX > 500);
        if (punishment >= PUNISHMENT_MAX)
            stats.banned++;
        else if (punishment >= 200)
            stats.partialBanned++;
        if (iter->second.address.ipAddress.is_v6())
            stats.ipv6Addresses++;
    }
    return stats;
}