/* vim: set expandtab ts=4 sw=4: */
/*
* You may redistribute this program 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 .
*/
#include "crypto/AddressCalc.h"
#include "crypto/Key.h"
#include "dht/dhtcore/ReplySerializer.h"
#include "subnode/SupernodeHunter.h"
#include "subnode/AddrSet.h"
#include "util/Identity.h"
#include "util/platform/Sockaddr.h"
#include "util/events/Timeout.h"
#include "util/AddrTools.h"
#include "util/events/Time.h"
#include "switch/LabelSplicer.h"
#include
#define CYCLE_MS 3000
struct SupernodeHunter_pvt
{
struct SupernodeHunter pub;
/** Nodes which are authorized to be our supernode. */
struct AddrSet* authorizedSnodes;
/** Our peers, DO NOT TOUCH, changed from in SubnodePathfinder. */
struct AddrSet* myPeerAddrs;
struct AddrSet* blacklist;
// Number of the next peer to ping in the peers AddrSet
int nextPeer;
// Will be set to the best known supernode possibility
struct Address snodeCandidate;
bool snodePathUpdated;
struct Allocator* alloc;
struct Log* log;
struct MsgCore* msgCore;
struct EventBase* base;
struct SwitchPinger* sp;
struct Address* myAddress;
String* selfAddrStr;
struct ReachabilityCollector* rc;
Identity
};
struct Query
{
struct SupernodeHunter_pvt* snp;
// If this is a findNode request, this is the search target, if it's a getPeers it's null.
struct Address* searchTar;
int64_t sendTime;
bool isGetRoute;
Identity
};
int SupernodeHunter_addSnode(struct SupernodeHunter* snh, struct Address* snodeAddr)
{
struct SupernodeHunter_pvt* snp = Identity_check((struct SupernodeHunter_pvt*) snh);
int length0 = snp->authorizedSnodes->length;
AddrSet_add(snp->authorizedSnodes, snodeAddr, AddrSet_Match_ADDRESS_ONLY);
if (snp->authorizedSnodes->length == length0) {
return SupernodeHunter_addSnode_EXISTS;
}
return 0;
}
int SupernodeHunter_listSnodes(struct SupernodeHunter* snh,
struct Address*** outP,
struct Allocator* alloc)
{
struct SupernodeHunter_pvt* snp = Identity_check((struct SupernodeHunter_pvt*) snh);
struct Address** out = Allocator_calloc(alloc, sizeof(char*), snp->authorizedSnodes->length);
for (int i = 0; i < snp->authorizedSnodes->length; i++) {
out[i] = AddrSet_get(snp->authorizedSnodes, i);
}
*outP = out;
return snp->authorizedSnodes->length;
}
int SupernodeHunter_removeSnode(struct SupernodeHunter* snh, struct Address* toRemove)
{
struct SupernodeHunter_pvt* snp = Identity_check((struct SupernodeHunter_pvt*) snh);
int length0 = snp->authorizedSnodes->length;
AddrSet_remove(snp->authorizedSnodes, toRemove, AddrSet_Match_ADDRESS_ONLY);
if (snp->authorizedSnodes->length == length0) {
return SupernodeHunter_removeSnode_NONEXISTANT;
}
return 0;
}
static struct Address* getPeerByNpn(struct SupernodeHunter_pvt* snp, int npn)
{
npn = npn % snp->myPeerAddrs->length;
int i = npn;
do {
struct Address* peer = AddrSet_get(snp->myPeerAddrs, i);
if (peer && peer->protocolVersion > 19) { return peer; }
i = (i + 1) % snp->myPeerAddrs->length;
} while (i != npn);
return NULL;
}
static void adoptSupernode2(Dict* msg, struct Address* src, struct MsgCore_Promise* prom)
{
struct Query* q = Identity_check((struct Query*) prom->userData);
struct SupernodeHunter_pvt* snp = Identity_check(q->snp);
if (!src) {
Log_debug(snp->log, "timeout sending to %s",
Address_toString(prom->target, prom->alloc)->bytes);
// If we're in this state and it doesn't work, we're going to drop the snode and
// go back to the beginning because while there's a possibility of a lost packet,
// it's a bigger possibility that we don't have a working path and we'd better
// try another one.
AddrSet_add(snp->blacklist, prom->target, AddrSet_Match_BOTH);
Bits_memset(&snp->snodeCandidate, 0, Address_SIZE);
snp->snodePathUpdated = false;
return;
}
Log_debug(snp->log, "Reply from %s", Address_toString(src, prom->alloc)->bytes);
int64_t* snodeRecvTime = Dict_getIntC(msg, "recvTime");
if (!snodeRecvTime) {
Log_info(snp->log, "getRoute reply with no timeStamp, bad snode");
return;
}
Log_debug(snp->log, "\n\nSupernode location confirmed [%s]\n\n",
Address_toString(src, prom->alloc)->bytes);
if (snp->pub.snodeIsReachable) {
// If while we were searching, the outside code declared that indeed the snode
// is reachable, we will not try to change their snode.
} else if (snp->pub.onSnodeChange) {
Bits_memcpy(&snp->pub.snodeAddr, src, Address_SIZE);
snp->pub.snodeIsReachable = (
AddrSet_indexOf(snp->authorizedSnodes, src, AddrSet_Match_ADDRESS_ONLY) != -1
) ? 2 : 1;
snp->pub.onSnodeChange(&snp->pub, q->sendTime, *snodeRecvTime);
} else {
Log_warn(snp->log, "onSnodeChange is not set");
}
}
static void adoptSupernode(struct SupernodeHunter_pvt* snp, struct Address* candidate)
{
struct MsgCore_Promise* qp = MsgCore_createQuery(snp->msgCore, 0, snp->alloc);
struct Query* q = Allocator_calloc(qp->alloc, sizeof(struct Query), 1);
Identity_set(q);
q->snp = snp;
q->sendTime = Time_currentTimeMilliseconds();
Dict* msg = qp->msg = Dict_new(qp->alloc);
qp->cb = adoptSupernode2;
qp->userData = q;
qp->target = Address_clone(candidate, qp->alloc);
// NOTE: we don't immediately request a path because the RS doesn't know about us
// quite yet, so it will tell us it doesn't know a path, so we need to ping it
// and take it on faith until we get some announcements announced.
Log_debug(snp->log, "Pinging snode [%s]", Address_toString(qp->target, qp->alloc)->bytes);
Dict_putStringCC(msg, "sq", "pn", qp->alloc);
Assert_true(AddressCalc_validAddress(candidate->ip6.bytes));
return;
}
static void updateSnodePath2(Dict* msg, struct Address* src, struct MsgCore_Promise* prom)
{
struct Query* q = Identity_check((struct Query*) prom->userData);
struct SupernodeHunter_pvt* snp = Identity_check(q->snp);
if (!src) {
String* addrStr = Address_toString(prom->target, prom->alloc);
Log_debug(snp->log, "timeout sending to %s", addrStr->bytes);
return;
}
int64_t* snodeRecvTime = Dict_getIntC(msg, "recvTime");
if (!snodeRecvTime) {
Log_info(snp->log, "getRoute reply with no timeStamp, bad snode");
return;
}
struct Address_List* al = ReplySerializer_parse(src, msg, snp->log, false, prom->alloc);
if (!al || al->length == 0) {
Log_debug(snp->log, "Requesting route to snode [%s], it doesn't know one",
Address_toString(prom->target, prom->alloc)->bytes);
return;
}
Log_debug(snp->log, "Supernode path updated with [%s]",
Address_toString(&al->elems[0], prom->alloc)->bytes);
snp->snodePathUpdated = true;
if (!Bits_memcmp(&snp->pub.snodeAddr, &al->elems[0], Address_SIZE)) {
Log_debug(snp->log, "Requestes route to snode [%s], the one we have is fine",
Address_toString(prom->target, prom->alloc)->bytes);
return;
}
Bits_memcpy(&snp->pub.snodeAddr, &al->elems[0], Address_SIZE);
Bits_memcpy(&snp->snodeCandidate, &al->elems[0], Address_SIZE);
AddrSet_flush(snp->blacklist);
if (snp->pub.onSnodeChange) {
snp->pub.snodeIsReachable = (
AddrSet_indexOf(snp->authorizedSnodes, src, AddrSet_Match_ADDRESS_ONLY) != -1
) ? 2 : 1;
snp->pub.onSnodeChange(&snp->pub, q->sendTime, *snodeRecvTime);
}
}
static void updateSnodePath(struct SupernodeHunter_pvt* snp)
{
struct MsgCore_Promise* qp = MsgCore_createQuery(snp->msgCore, 0, snp->alloc);
struct Query* q = Allocator_calloc(qp->alloc, sizeof(struct Query), 1);
Identity_set(q);
q->snp = snp;
q->sendTime = Time_currentTimeMilliseconds();
Dict* msg = qp->msg = Dict_new(qp->alloc);
qp->cb = updateSnodePath2;
qp->userData = q;
qp->target = Address_clone(&snp->pub.snodeAddr, qp->alloc);
Log_debug(snp->log, "Update snode [%s] path", Address_toString(qp->target, qp->alloc)->bytes);
Dict_putStringCC(msg, "sq", "gr", qp->alloc);
String* src = String_newBinary(snp->myAddress->ip6.bytes, 16, qp->alloc);
Dict_putStringC(msg, "src", src, qp->alloc);
String* target = String_newBinary(snp->pub.snodeAddr.ip6.bytes, 16, qp->alloc);
Dict_putStringC(msg, "tar", target, qp->alloc);
}
static void queryForAuthorized(struct SupernodeHunter_pvt* snp, struct Address* snode)
{
/*
struct MsgCore_Promise* qp = MsgCore_createQuery(snp->msgCore, 0, snp->alloc);
struct Query* q = Allocator_calloc(qp->alloc, sizeof(struct Query), 1);
Identity_set(q);
q->snp = snp;
q->sendTime = Time_currentTimeMilliseconds();
Dict* msg = qp->msg = Dict_new(qp->alloc);
qp->cb = onReply;
qp->userData = q;
qp->target = candidate;
Log_debug(snp->log, "Pinging snode [%s]", Address_toString(qp->target, qp->alloc)->bytes);
Dict_putStringCC(msg, "sq", "gr", qp->alloc);
*/
}
static void peerResponseOK(struct SwitchPinger_Response* resp, struct SupernodeHunter_pvt* snp)
{
ReachabilityCollector_lagSample(snp->rc, resp->label, resp->milliseconds);
struct Address snode = {0};
Bits_memcpy(&snode, &resp->snode, sizeof(struct Address));
if (!snode.path) {
uint8_t label[20];
AddrTools_printPath(label, resp->label);
Log_debug(snp->log, "Peer [%s] reports no supernode", label);
return;
}
uint64_t path = LabelSplicer_splice(snode.path, resp->label);
if (path == UINT64_MAX) {
Log_debug(snp->log, "Supernode path could not be spliced");
return;
}
snode.path = path;
struct Address peerAddr = { .path = resp->label };
int i = AddrSet_indexOf(snp->myPeerAddrs, &peerAddr, AddrSet_Match_LABEL_ONLY);
if (i == -1) {
Log_info(snp->log, "We got a snode reply from a node which is not in peer list");
return;
}
struct Address* peer = AddrSet_get(snp->myPeerAddrs, i);
struct Address* firstPeer = getPeerByNpn(snp, 0);
if (!firstPeer) {
Log_info(snp->log, "All peers have gone away while packet was outstanding");
return;
}
// 1.
// If we have looped around and queried all of our peers returning to the first and we have
// still not found an snode in our authorized snodes list, we should simply accept this one.
if (!snp->pub.snodeIsReachable &&
snp->myPeerAddrs->length > 1 &&
snp->nextPeer >= snp->myPeerAddrs->length &&
Address_isSameIp(firstPeer, peer))
{
if (!snp->snodeCandidate.path) {
Log_info(snp->log, "No snode candidate found [%s]",
Address_toStringKey(&snp->snodeCandidate, resp->ping->pingAlloc)->bytes);
snp->nextPeer = 0;
AddrSet_flush(snp->blacklist);
return;
}
Log_debug(snp->log, "Peer [%s] has proposed we use supernode [%s] we will accept it",
Address_toString(peer, resp->ping->pingAlloc)->bytes,
Address_toString(&snp->snodeCandidate, resp->ping->pingAlloc)->bytes);
adoptSupernode(snp, &snp->snodeCandidate);
return;
}
// 2.
// If this snode is one of our authorized snodes OR if we have none defined, accept this one.
if (AddrSet_indexOf(snp->blacklist, &snode, AddrSet_Match_BOTH) > -1) {
Log_debug(snp->log, "Peer [%s] [%" PRIx64 "] has proposed supernode [%s] "
"but it is blacklisted, continue",
Address_toString(peer, resp->ping->pingAlloc)->bytes,
resp->label,
Address_toString(&snode, resp->ping->pingAlloc)->bytes);
} else if (!snp->authorizedSnodes->length ||
AddrSet_indexOf(snp->authorizedSnodes, &snode, AddrSet_Match_ADDRESS_ONLY) > -1)
{
Address_getPrefix(&snode);
Log_debug(snp->log, "Peer [%s] has proposed supernode [%s] and %s so we will use it",
Address_toString(peer, resp->ping->pingAlloc)->bytes,
Address_toString(&snode, resp->ping->pingAlloc)->bytes,
(snp->authorizedSnodes->length) ? "it is authorized" : "we have none authorized");
adoptSupernode(snp, &snode);
return;
} else if (!snp->snodeCandidate.path) {
Log_debug(snp->log, "Peer [%s] has proposed supernode [%s], we're not using it yet "
"but we will store it as a candidate.",
Address_toString(peer, resp->ping->pingAlloc)->bytes,
Address_toString(&snp->snodeCandidate, resp->ping->pingAlloc)->bytes);
Bits_memcpy(&snp->snodeCandidate, &snode, sizeof(struct Address));
Address_getPrefix(&snp->snodeCandidate);
}
// 3.
// If this snode is not one of our authorized snodes, query it for all of our authorized snodes.
queryForAuthorized(snp, &snode);
}
static void peerResponse(struct SwitchPinger_Response* resp, void* userData)
{
struct SupernodeHunter_pvt* snp = Identity_check((struct SupernodeHunter_pvt*) userData);
char* err = "";
switch (resp->res) {
case SwitchPinger_Result_OK: peerResponseOK(resp, snp); return;
case SwitchPinger_Result_LABEL_MISMATCH: err = "LABEL_MISMATCH"; break;
case SwitchPinger_Result_WRONG_DATA: err = "WRONG_DATA"; break;
case SwitchPinger_Result_ERROR_RESPONSE: err = "ERROR_RESPONSE"; break;
case SwitchPinger_Result_LOOP_ROUTE: err = "LOOP_ROUTE"; break;
case SwitchPinger_Result_TIMEOUT: err = "TIMEOUT"; break;
default: err = "unknown error"; break;
}
Log_debug(snp->log, "Error sending snp query to peer [%" PRIx64 "] [%s]",
resp->label, err);
}
static void probePeerCycle(void* vsn)
{
struct SupernodeHunter_pvt* snp = Identity_check((struct SupernodeHunter_pvt*) vsn);
if (snp->pub.snodeIsReachable && !snp->snodePathUpdated) {
updateSnodePath(snp);
}
if (snp->pub.snodeIsReachable > 1) { return; }
if (snp->pub.snodeIsReachable && !snp->authorizedSnodes->length) { return; }
if (!snp->myPeerAddrs->length) { return; }
//Log_debug(snp->log, "probePeerCycle()");
if (AddrSet_indexOf(snp->authorizedSnodes, snp->myAddress, AddrSet_Match_ADDRESS_ONLY) != -1) {
Log_info(snp->log, "Self is specified as supernode, pinging...");
adoptSupernode(snp, snp->myAddress);
return;
}
struct Address* peer = getPeerByNpn(snp, snp->nextPeer);
if (!peer) {
Log_info(snp->log, "No peer found who is version >= 20");
return;
}
struct SwitchPinger_Ping* p =
SwitchPinger_newPing(peer->path, String_CONST(""), 3000, peerResponse, snp->alloc, snp->sp);
Assert_true(p);
Log_debug(snp->log, "Querying peer [%s] [%d] total [%d], blacklist size [%d]",
Address_toString(peer, p->pingAlloc)->bytes,
snp->nextPeer,
snp->myPeerAddrs->length,
snp->blacklist->length);
snp->nextPeer++;
p->type = SwitchPinger_Type_GETSNODE;
if (snp->pub.snodeIsReachable) {
Bits_memcpy(&p->snode, &snp->pub.snodeAddr, sizeof(struct Address));
}
p->onResponseContext = snp;
}
static void onSnodeUnreachable(struct SupernodeHunter* snh,
int64_t sendTime,
int64_t snodeRecvTime)
{
struct SupernodeHunter_pvt* snp = Identity_check((struct SupernodeHunter_pvt*) snh);
Log_debug(snp->log, "Supernode unreachable.");
snp->snodePathUpdated = false;
// Snode unreachable, we need also reset peer snode candidate
Bits_memset(&snp->snodeCandidate, 0, Address_SIZE);
}
struct SupernodeHunter* SupernodeHunter_new(struct Allocator* allocator,
struct Log* log,
struct EventBase* base,
struct SwitchPinger* sp,
struct AddrSet* peers,
struct MsgCore* msgCore,
struct Address* myAddress,
struct ReachabilityCollector* rc)
{
struct Allocator* alloc = Allocator_child(allocator);
struct SupernodeHunter_pvt* out =
Allocator_calloc(alloc, sizeof(struct SupernodeHunter_pvt), 1);
Identity_set(out);
out->authorizedSnodes = AddrSet_new(alloc);
out->blacklist = AddrSet_new(alloc);
out->myPeerAddrs = peers;
out->base = base;
//out->rand = rand;
//out->nodes = AddrSet_new(alloc);
//out->timeSnodeCalled = Time_currentTimeMilliseconds();
//out->snodeCandidates = AddrSet_new(alloc);
out->log = log;
out->alloc = alloc;
out->msgCore = msgCore;
out->myAddress = myAddress;
out->rc = rc;
out->selfAddrStr = String_newBinary(myAddress->ip6.bytes, 16, alloc);
out->sp = sp;
out->snodePathUpdated = false;
out->pub.onSnodeUnreachable = onSnodeUnreachable;
Timeout_setInterval(probePeerCycle, out, CYCLE_MS, base, alloc);
return &out->pub;
}