network_engine.h

/*
 *  Copyright (C) 2014-2022 Savoir-faire Linux Inc.
 *  Author(s) : Adrien Béraud <adrien.beraud@savoirfairelinux.com>
 *              Simon Désaulniers <simon.desaulniers@savoirfairelinux.com>
 *
 *  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 <https://www.gnu.org/licenses/>.
 */

#pragma once

#include "node_cache.h"
#include "value.h"
#include "infohash.h"
#include "node.h"
#include "scheduler.h"
#include "utils.h"
#include "rng.h"
#include "rate_limiter.h"
#include "log_enable.h"
#include "network_utils.h"

#include <vector>
#include <string>
#include <functional>
#include <algorithm>
#include <memory>
#include <queue>

namespace dht {
namespace net {

struct Request;
struct Socket;
struct TransId;

#ifndef MSG_CONFIRM
#define MSG_CONFIRM 0
#endif

struct NetworkConfig {
    NetId network {0};
    ssize_t max_req_per_sec {0};
    ssize_t max_peer_req_per_sec {0};
};

class DhtProtocolException : public DhtException {
public:
    // sent to another peer (http-like).
    static const constexpr uint16_t NON_AUTHORITATIVE_INFORMATION {203}; /* incomplete request packet. */
    static const constexpr uint16_t UNAUTHORIZED {401};                  /* wrong tokens. */
    static const constexpr uint16_t NOT_FOUND {404};                     /* storage not found */
    // for internal use (custom).
    static const constexpr uint16_t INVALID_TID_SIZE {421};              /* id was truncated. */
    static const constexpr uint16_t UNKNOWN_TID {422};                   /* unknown tid */
    static const constexpr uint16_t WRONG_NODE_INFO_BUF_LEN {423};       /* node info length is wrong */

    static const std::string GET_NO_INFOHASH;    /* received "get" request with no infohash */
    static const std::string LISTEN_NO_INFOHASH; /* got "listen" request without infohash */
    static const std::string LISTEN_WRONG_TOKEN; /* wrong token in "listen" request */
    static const std::string PUT_NO_INFOHASH;    /* no infohash in "put" request */
    static const std::string PUT_WRONG_TOKEN;    /* got "put" request with wrong token */
    static const std::string STORAGE_NOT_FOUND;  /* got access request for an unknown storage */
    static const std::string PUT_INVALID_ID;     /* invalid id in "put" request */

    DhtProtocolException(uint16_t code, const std::string& msg="", InfoHash failing_node_id={})
        : DhtException(msg), msg(msg), code(code), failing_node_id(failing_node_id) {}

    const std::string& getMsg() const { return msg; }
    uint16_t getCode() const { return code; }
    const InfoHash& getNodeId() const { return failing_node_id; }

private:
    std::string msg;
    uint16_t code;
    InfoHash failing_node_id;
};

struct ParsedMessage;

struct RequestAnswer {
    Blob ntoken {};
    Value::Id vid {};
    std::vector<Sp<Value>> values {};
    std::vector<Value::Id> refreshed_values {};
    std::vector<Value::Id> expired_values {};
    std::vector<Sp<FieldValueIndex>> fields {};
    std::vector<Sp<Node>> nodes4 {};
    std::vector<Sp<Node>> nodes6 {};
    RequestAnswer() {}
    RequestAnswer(ParsedMessage&& msg);
};

class NetworkEngine final
{
private:
    std::function<void(Sp<Request>, DhtProtocolException)> onError;

    std::function<void(const Sp<Node>&, int)> onNewNode;
    std::function<void(const InfoHash&, const SockAddr&)> onReportedAddr;
    std::function<RequestAnswer(Sp<Node>)> onPing {};
    std::function<RequestAnswer(Sp<Node>, const InfoHash&, want_t)> onFindNode {};
    std::function<RequestAnswer(Sp<Node>, const InfoHash&, want_t, const Query&)> onGetValues {};
    std::function<RequestAnswer(Sp<Node>,
            const InfoHash&,
            const Blob&,
            Tid,
            const Query&,
            int)> onListen {};
    std::function<RequestAnswer(Sp<Node>,
            const InfoHash&,
            const Blob&,
            const std::vector<Sp<Value>>&,
            const time_point&)> onAnnounce {};
    std::function<RequestAnswer(Sp<Node>,
            const InfoHash&,
            const Blob&,
            const Value::Id&)> onRefresh {};

public:
    using RequestCb = std::function<void(const Request&, RequestAnswer&&)>;
    using RequestErrorCb = std::function<bool(const Request&, DhtProtocolException&&)>;
    using RequestExpiredCb = std::function<void(const Request&, bool)>;

    NetworkEngine(
            InfoHash& myid,
            NetworkConfig config,
            std::unique_ptr<DatagramSocket>&& sock,
            const Sp<Logger>& log,
            std::mt19937_64& rd,
            Scheduler& scheduler,
            decltype(NetworkEngine::onError)&& onError,
            decltype(NetworkEngine::onNewNode)&& onNewNode,
            decltype(NetworkEngine::onReportedAddr)&& onReportedAddr,
            decltype(NetworkEngine::onPing)&& onPing,
            decltype(NetworkEngine::onFindNode)&& onFindNode,
            decltype(NetworkEngine::onGetValues)&& onGetValues,
            decltype(NetworkEngine::onListen)&& onListen,
            decltype(NetworkEngine::onAnnounce)&& onAnnounce,
            decltype(NetworkEngine::onRefresh)&& onRefresh);

    ~NetworkEngine();

    net::DatagramSocket* getSocket() const { return dht_socket.get(); };

    void clear();

    void tellListener(const Sp<Node>& n, Tid socket_id, const InfoHash& hash, want_t want, const Blob& ntoken,
            std::vector<Sp<Node>>&& nodes, std::vector<Sp<Node>>&& nodes6,
            std::vector<Sp<Value>>&& values, const Query& q, int version);

    void tellListenerRefreshed(const Sp<Node>& n, Tid socket_id, const InfoHash& hash, const Blob& ntoken, const std::vector<Value::Id>& values, int version);
    void tellListenerExpired(const Sp<Node>& n, Tid socket_id, const InfoHash& hash, const Blob& ntoken, const std::vector<Value::Id>& values, int version);

    bool isRunning(sa_family_t af) const;
    inline want_t want () const { return dht_socket->hasIPv4() and dht_socket->hasIPv6() ? (WANT4 | WANT6) : -1; }

    void connectivityChanged(sa_family_t);

    /**************
     *  Requests  *
     **************/

    Sp<Request>
    sendPing(const Sp<Node>& n, RequestCb&& on_done, RequestExpiredCb&& on_expired);

    Sp<Request>
    sendPing(SockAddr&& sa, RequestCb&& on_done, RequestExpiredCb&& on_expired) {
        return sendPing(std::make_shared<Node>(InfoHash::zero(), std::move(sa), rd),
                std::forward<RequestCb>(on_done),
                std::forward<RequestExpiredCb>(on_expired));
    }

    Sp<Request> sendFindNode(const Sp<Node>& n,
                             const InfoHash& hash,
                             want_t want = -1,
                             RequestCb&& on_done = {},
                             RequestExpiredCb&& on_expired = {});
    Sp<Request> sendGetValues(const Sp<Node>& n,
                              const InfoHash& hash,
                              const Query& query,
                              want_t want,
                              RequestCb&& on_done,
                              RequestExpiredCb&& on_expired);
    Sp<Request> sendListen(const Sp<Node>& n,
                           const InfoHash& hash,
                           const Query& query,
                           const Blob& token,
                           Tid socketId,
                           RequestCb&& on_done,
                           RequestExpiredCb&& on_expired);
    Sp<Request> sendAnnounceValue(const Sp<Node>& n,
                                  const InfoHash& hash,
                                  const Sp<Value>& v,
                                  time_point created,
                                  const Blob& token,
                                  RequestCb&& on_done,
                                  RequestExpiredCb&& on_expired);
    Sp<Request> sendRefreshValue(const Sp<Node>& n,
                                 const InfoHash& hash,
                                 const Value::Id& vid,
                                 const Blob& token,
                                 RequestCb&& on_done,
                                 RequestErrorCb&& on_error,
                                 RequestExpiredCb&& on_expired);
    Sp<Request> sendUpdateValues(const Sp<Node>& n,
                                 const InfoHash& infohash,
                                 const std::vector<Sp<Value>>& values,
                                 time_point created,
                                 const Blob& token,
                                 const size_t& sid);

    void processMessage(const uint8_t *buf, size_t buflen, SockAddr addr);

    Sp<Node> insertNode(const InfoHash& id, const SockAddr& addr) {
        auto n = cache.getNode(id, addr, scheduler.time(), 0);
        onNewNode(n, 0);
        return n;
    }

    std::vector<unsigned> getNodeMessageStats(bool in) {
        auto& st = in ? in_stats : out_stats;
        std::vector<unsigned> stats {st.ping,  st.find,  st.get,  st.listen,  st.put};
        st = {};
        return stats;
    }

    void blacklistNode(const Sp<Node>& n);

    std::vector<Sp<Node>> getCachedNodes(const InfoHash& id, sa_family_t sa_f, size_t count) {
        return cache.getCachedNodes(id, sa_f, count);
    }

    size_t getNodeCacheSize() const {
        return cache.size();
    }
    size_t getNodeCacheSize(sa_family_t af) const {
        return cache.size(af);
    }

    size_t getRateLimiterSize() const {
        return address_rate_limiter.size();
    }

    size_t getPartialCount() const {
        return partial_messages.size();
    }

private:

    struct PartialMessage;

    /***************
     *  Constants  *
     ***************/
    /* the length of a node info buffer in ipv4 format */
    static const constexpr size_t NODE4_INFO_BUF_LEN {HASH_LEN + sizeof(in_addr) + sizeof(in_port_t)};
    /* the length of a node info buffer in ipv6 format */
    static const constexpr size_t NODE6_INFO_BUF_LEN {HASH_LEN + sizeof(in6_addr) + sizeof(in_port_t)};
    /* after a UDP reply, the period during which we tell the link layer about it */
    static constexpr std::chrono::seconds UDP_REPLY_TIME {15};

    /* Max. time to receive a full fragmented packet */
    static constexpr std::chrono::seconds RX_MAX_PACKET_TIME {10};
    /* Max. time between packet fragments */
    static constexpr std::chrono::seconds RX_TIMEOUT {3};
    /* The maximum number of nodes that we snub.  There is probably little
        reason to increase this value. */
    static constexpr unsigned BLACKLISTED_MAX {10};

    static constexpr size_t MTU {1280};
    static constexpr size_t MAX_PACKET_VALUE_SIZE {600};

    static const std::string my_v;

    void process(std::unique_ptr<ParsedMessage>&&, const SockAddr& from);

    bool rateLimit(const SockAddr& addr);

    static bool isMartian(const SockAddr& addr);
    bool isNodeBlacklisted(const SockAddr& addr) const;

    void requestStep(Sp<Request> req);

    void sendRequest(const Sp<Request>& request);

    struct MessageStats {
        unsigned ping         {0};
        unsigned find         {0};
        unsigned get          {0};
        unsigned put          {0};
        unsigned listen       {0};
        unsigned refresh      {0};
        unsigned updateValue  {0};
    };


    // basic wrapper for socket sendto function
    int send(const SockAddr& addr, const char *buf, size_t len, bool confirmed = false);

    void sendValueParts(Tid tid, const std::vector<Blob>& svals, const SockAddr& addr);
    std::vector<Blob> packValueHeader(msgpack::sbuffer&, const std::vector<Sp<Value>>&);
    void maintainRxBuffer(Tid tid);

    /*************
     *  Answers  *
     *************/
    /* answer to a ping  request */
    void sendPong(const SockAddr& addr, Tid tid);
    /* answer to findnodes/getvalues request */
    void sendNodesValues(const SockAddr& addr,
            Tid tid,
            const Blob& nodes,
            const Blob& nodes6,
            const std::vector<Sp<Value>>& st,
            const Query& query,
            const Blob& token);
    Blob bufferNodes(sa_family_t af, const InfoHash& id, std::vector<Sp<Node>>& nodes);

    std::pair<Blob, Blob> bufferNodes(sa_family_t af,
            const InfoHash& id,
            want_t want,
            std::vector<Sp<Node>>& nodes,
            std::vector<Sp<Node>>& nodes6);
    /* answer to a listen request */
    void sendListenConfirmation(const SockAddr& addr, Tid tid);
    /* answer to put request */
    void sendValueAnnounced(const SockAddr& addr, Tid, Value::Id);
    /* answer in case of error */
    void sendError(const SockAddr& addr,
            Tid tid,
            uint16_t code,
            const std::string& message,
            bool include_id=false);

    void deserializeNodes(ParsedMessage& msg, const SockAddr& from);

    /* DHT info */
    const InfoHash& myid;
    const NetworkConfig config {};
    const std::unique_ptr<DatagramSocket> dht_socket;
    Sp<Logger> logger_;
    std::mt19937_64& rd;

    NodeCache cache;

    // global limiting should be triggered by at least 8 different IPs
    using IpLimiter = RateLimiter;
    using IpLimiterMap = std::map<SockAddr, IpLimiter, SockAddr::ipCmp>;
    IpLimiterMap address_rate_limiter;
    RateLimiter rate_limiter;
    ssize_t limiter_maintenance {0};

    // requests handling
    std::map<Tid, Sp<Request>> requests {};
    std::map<Tid, PartialMessage> partial_messages;

    MessageStats in_stats {}, out_stats {};
    std::set<SockAddr> blacklist {};

    Scheduler& scheduler;

    bool logIncoming_ {false};
};

} /* namespace net  */
} /* namespace dht */