/* 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 "dht/Address.h"
#include "net/ControlHandler.h"
#include "util/Identity.h"
#include "util/AddrTools.h"
#include "util/Checksum.h"
#include "wire/Control.h"
#include "wire/Error.h"
#define NumberCompress_OLD_CODE
#include "switch/NumberCompress.h"
#include
struct ControlHandler_pvt
{
struct ControlHandler pub;
struct Log* log;
struct Allocator* alloc;
uint8_t myPublicKey[32];
struct Iface eventIf;
struct Address activeSnode;
Identity
};
/**
* Expects [ Ctrl ][ Error ][ cause SwitchHeader ][ cause handle ][ cause etc.... ]
*/
#define handleError_MIN_SIZE (Control_Header_SIZE + Control_Error_MIN_SIZE + SwitchHeader_SIZE + 4)
static Iface_DEFUN handleError(struct Message* msg,
struct ControlHandler_pvt* ch,
uint64_t label,
uint8_t* labelStr,
struct RouteHeader* rh)
{
if (Message_getLength(msg) < handleError_MIN_SIZE) {
Log_info(ch->log, "DROP runt error packet from [%s]", labelStr);
return Error(msg, "RUNT");
}
Er_assert(Message_truncate(msg, handleError_MIN_SIZE));
Er_assert(Message_epush(msg, &rh->sh, SwitchHeader_SIZE));
Er_assert(Message_epush32be(msg, 0xffffffff));
Er_assert(Message_epush32be(msg, PFChan_Core_SWITCH_ERR));
return Iface_next(&ch->eventIf, msg);
}
/**
* Expects [ SwitchHeader ][ Ctrl ][ (key)Ping ][ data etc.... ]
*/
#define handlePing_MIN_SIZE (Control_Header_SIZE + Control_Ping_MIN_SIZE)
static Iface_DEFUN handlePing(struct Message* msg,
struct ControlHandler_pvt* ch,
uint64_t label,
uint8_t* labelStr,
uint16_t messageType_be)
{
if (Message_getLength(msg) < handlePing_MIN_SIZE) {
Log_info(ch->log, "DROP runt ping");
return Error(msg, "RUNT");
}
struct Control* ctrl = (struct Control*) msg->msgbytes;
Er_assert(Message_eshift(msg, -Control_Header_SIZE));
// Ping and keyPing share version location
struct Control_Ping* ping = (struct Control_Ping*) msg->msgbytes;
uint32_t herVersion = Endian_bigEndianToHost32(ping->version_be);
if (!Version_isCompatible(Version_CURRENT_PROTOCOL, herVersion)) {
Log_debug(ch->log, "DROP ping from incompatible version [%d]", herVersion);
return Error(msg, "INVALID");
}
if (messageType_be == Control_KEYPING_be) {
Log_debug(ch->log, "got switch keyPing from [%s]", labelStr);
if (Message_getLength(msg) < Control_KeyPing_HEADER_SIZE) {
// min keyPing size is longer
Log_debug(ch->log, "DROP runt keyPing");
return Error(msg, "RUNT");
}
if (Message_getLength(msg) > Control_KeyPing_MAX_SIZE) {
Log_debug(ch->log, "DROP long keyPing");
return Error(msg, "INVALID");
}
if (ping->magic != Control_KeyPing_MAGIC) {
Log_debug(ch->log, "DROP keyPing (bad magic)");
return Error(msg, "INVALID");
}
struct Control_KeyPing* keyPing = (struct Control_KeyPing*) msg->msgbytes;
keyPing->magic = Control_KeyPong_MAGIC;
ctrl->header.type_be = Control_KEYPONG_be;
Bits_memcpy(keyPing->key, ch->myPublicKey, 32);
} else if (messageType_be == Control_PING_be) {
// Happens in benchmark.
//Log_debug(ch->log, "got switch ping from [%s]", labelStr);
if (ping->magic != Control_Ping_MAGIC) {
Log_debug(ch->log, "DROP ping (bad magic)");
return Error(msg, "INVALID");
}
ping->magic = Control_Pong_MAGIC;
ctrl->header.type_be = Control_PONG_be;
} else {
Assert_failure("2+2=5");
}
ping->version_be = Endian_hostToBigEndian32(Version_CURRENT_PROTOCOL);
Er_assert(Message_eshift(msg, Control_Header_SIZE));
ctrl->header.checksum_be = 0;
ctrl->header.checksum_be = Checksum_engine_be(msg->msgbytes, Message_getLength(msg));
Er_assert(Message_eshift(msg, RouteHeader_SIZE));
struct RouteHeader* routeHeader = (struct RouteHeader*) msg->msgbytes;
Bits_memset(routeHeader, 0, RouteHeader_SIZE);
SwitchHeader_setVersion(&routeHeader->sh, SwitchHeader_CURRENT_VERSION);
routeHeader->sh.label_be = Endian_hostToBigEndian64(label);
routeHeader->flags |= RouteHeader_flags_CTRLMSG;
return Iface_next(&ch->pub.coreIf, msg);
}
/**
* Expects [ SwitchHeader ][ Ctrl ][ RPath ][ data etc.... ]
*/
#define handleRPathQuery_MIN_SIZE (Control_Header_SIZE + Control_RPath_HEADER_SIZE)
static Iface_DEFUN handleRPathQuery(struct Message* msg,
struct ControlHandler_pvt* ch,
uint64_t label,
uint8_t* labelStr)
{
Log_debug(ch->log, "Incoming RPATH query");
if (Message_getLength(msg) < handleRPathQuery_MIN_SIZE) {
Log_info(ch->log, "DROP runt RPATH query");
return Error(msg, "RUNT");
}
struct Control* ctrl = (struct Control*) msg->msgbytes;
struct Control_RPath* rpa = &ctrl->content.rpath;
if (rpa->magic != Control_RPATH_QUERY_MAGIC) {
Log_debug(ch->log, "DROP RPATH query (bad magic)");
return Error(msg, "INVALID");
}
ctrl->header.type_be = Control_RPATH_REPLY_be;
rpa->version_be = Endian_hostToBigEndian32(Version_CURRENT_PROTOCOL);
rpa->magic = Control_RPATH_REPLY_MAGIC;
uint64_t label_be = Endian_hostToBigEndian64(label);
Bits_memcpy(rpa->rpath_be, &label_be, 8);
ctrl->header.checksum_be = 0;
ctrl->header.checksum_be = Checksum_engine_be(msg->msgbytes, Message_getLength(msg));
Er_assert(Message_eshift(msg, RouteHeader_SIZE));
struct RouteHeader* routeHeader = (struct RouteHeader*) msg->msgbytes;
Bits_memset(routeHeader, 0, RouteHeader_SIZE);
SwitchHeader_setVersion(&routeHeader->sh, SwitchHeader_CURRENT_VERSION);
routeHeader->sh.label_be = label_be;
routeHeader->flags |= RouteHeader_flags_CTRLMSG;
return Iface_next(&ch->pub.coreIf, msg);
}
/**
* Expects [ SwitchHeader ][ Ctrl ][ SupernodeQuery ][ data etc.... ]
*/
#define handleGetSnodeQuery_MIN_SIZE (Control_Header_SIZE + Control_GetSnode_HEADER_SIZE)
static Iface_DEFUN handleGetSnodeQuery(struct Message* msg,
struct ControlHandler_pvt* ch,
uint64_t label,
uint8_t* labelStr)
{
Log_debug(ch->log, "incoming getSupernode query");
if (Message_getLength(msg) < handleGetSnodeQuery_MIN_SIZE) {
Log_info(ch->log, "DROP runt getSupernode query");
return Error(msg, "RUNT");
}
struct Control* ctrl = (struct Control*) msg->msgbytes;
struct Control_GetSnode* snq = &ctrl->content.getSnode;
if (snq->magic != Control_GETSNODE_QUERY_MAGIC) {
Log_debug(ch->log, "DROP getSupernode query (bad magic)");
return Error(msg, "INVALID");
}
uint32_t herVersion = Endian_bigEndianToHost32(snq->version_be);
if (!Version_isCompatible(Version_CURRENT_PROTOCOL, herVersion)) {
Log_debug(ch->log, "DROP getSupernode query from incompatible version [%d]", herVersion);
// Nothing wrong with the query but we're just not going to answer it
return NULL;
}
ctrl->header.type_be = Control_GETSNODE_REPLY_be;
snq->kbps_be = 0xffffffff;
snq->version_be = Endian_hostToBigEndian32(Version_CURRENT_PROTOCOL);
snq->magic = Control_GETSNODE_REPLY_MAGIC;
uint64_t fixedLabel = 0;
if (ch->activeSnode.path) {
fixedLabel = NumberCompress_getLabelFor(ch->activeSnode.path, label);
uint64_t fixedLabel_be = Endian_hostToBigEndian64(fixedLabel);
Bits_memcpy(snq->pathToSnode_be, &fixedLabel_be, 8);
Bits_memcpy(snq->snodeKey, ch->activeSnode.key, 32);
snq->snodeVersion_be = Endian_hostToBigEndian32(ch->activeSnode.protocolVersion);
} else {
Log_debug(ch->log, "getSupernode query and we do not have a known snode");
snq->snodeVersion_be = 0;
Bits_memset(snq->pathToSnode_be, 0, 8);
Bits_memset(snq->snodeKey, 0, 32);
}
ctrl->header.checksum_be = 0;
ctrl->header.checksum_be = Checksum_engine_be(msg->msgbytes, Message_getLength(msg));
Er_assert(Message_eshift(msg, RouteHeader_SIZE));
struct RouteHeader* routeHeader = (struct RouteHeader*) msg->msgbytes;
Bits_memset(routeHeader, 0, RouteHeader_SIZE);
SwitchHeader_setVersion(&routeHeader->sh, SwitchHeader_CURRENT_VERSION);
routeHeader->sh.label_be = Endian_hostToBigEndian64(label);
routeHeader->flags |= RouteHeader_flags_CTRLMSG;
Log_debug(ch->log, "Replied to [%" PRIx64 "] with snode [%" PRIx64 "] aka [%" PRIx64 "]",
label, fixedLabel, ch->activeSnode.path);
return Iface_next(&ch->pub.coreIf, msg);
}
/**
* Handle an incoming control message from a switch.
*
* @param context the ducttape context.
* @param message the control message, this should be alligned on the beginning of the content,
* that is to say, after the end of the switch header.
* @param switchHeader the header.
* @param switchIf the interface which leads to the switch.
* @param isFormV8 true if the control message is in the form specified by protocol version 8+
*/
static Iface_DEFUN incomingFromCore(struct Message* msg, struct Iface* coreIf)
{
struct ControlHandler_pvt* ch = Identity_check((struct ControlHandler_pvt*) coreIf);
struct RouteHeader routeHdr;
Er_assert(Message_epop(msg, &routeHdr, RouteHeader_SIZE));
uint8_t labelStr[20];
uint64_t label = Endian_bigEndianToHost64(routeHdr.sh.label_be);
AddrTools_printPath(labelStr, label);
// happens in benchmark
// Log_debug(ch->log, "ctrl packet from [%s]", labelStr);
if (Message_getLength(msg) < 4 + Control_Header_SIZE) {
Log_info(ch->log, "DROP runt ctrl packet from [%s]", labelStr);
return Error(msg, "RUNT");
}
Assert_true(routeHdr.flags & RouteHeader_flags_CTRLMSG);
if (Checksum_engine_be(msg->msgbytes, Message_getLength(msg))) {
Log_info(ch->log, "DROP ctrl packet from [%s] with invalid checksum", labelStr);
return Error(msg, "INVALID");
}
struct Control* ctrl = (struct Control*) msg->msgbytes;
if (ctrl->header.type_be == Control_ERROR_be) {
return handleError(msg, ch, label, labelStr, &routeHdr);
} else if (ctrl->header.type_be == Control_KEYPING_be
|| ctrl->header.type_be == Control_PING_be)
{
return handlePing(msg, ch, label, labelStr, ctrl->header.type_be);
} else if (ctrl->header.type_be == Control_KEYPONG_be
|| ctrl->header.type_be == Control_PONG_be)
{
Log_debug(ch->log, "got switch pong from [%s]", labelStr);
Er_assert(Message_epush(msg, &routeHdr, RouteHeader_SIZE));
return Iface_next(&ch->pub.switchPingerIf, msg);
} else if (ctrl->header.type_be == Control_GETSNODE_QUERY_be) {
return handleGetSnodeQuery(msg, ch, label, labelStr);
} else if (ctrl->header.type_be == Control_GETSNODE_REPLY_be
|| ctrl->header.type_be == Control_RPATH_REPLY_be)
{
Log_debug(ch->log, "got %s REPLY from [%s]",
(ctrl->header.type_be == Control_GETSNODE_REPLY_be) ? "GETSNODE" : "RPATH",
labelStr);
Er_assert(Message_epush(msg, &routeHdr, RouteHeader_SIZE));
Er_assert(Message_epush32be(msg, 0xffffffff));
Er_assert(Message_epush32be(msg, PFChan_Core_CTRL_MSG));
return Iface_next(&ch->eventIf, msg);
} else if (ctrl->header.type_be == Control_RPATH_QUERY_be) {
return handleRPathQuery(msg, ch, label, labelStr);
}
Log_info(ch->log, "DROP control packet of unknown type from [%s], type [%d]",
labelStr, Endian_bigEndianToHost16(ctrl->header.type_be));
return Error(msg, "INVALID");
}
// Forward from switch pinger directly to core.
static Iface_DEFUN incomingFromSwitchPinger(struct Message* msg, struct Iface* switchPingerIf)
{
struct ControlHandler_pvt* ch =
Identity_containerOf(switchPingerIf, struct ControlHandler_pvt, pub.switchPingerIf);
return Iface_next(&ch->pub.coreIf, msg);
}
static Iface_DEFUN changeSnode(struct Message* msg, struct Iface* eventIf)
{
struct ControlHandler_pvt* ch =
Identity_containerOf(eventIf, struct ControlHandler_pvt, eventIf);
enum PFChan_Pathfinder ev = Er_assert(Message_epop32be(msg));
Assert_true(ev == PFChan_Pathfinder_SNODE);
Er_assert(Message_epop32be(msg));
struct PFChan_Node node;
Er_assert(Message_epop(msg, &node, PFChan_Node_SIZE));
Assert_true(!Message_getLength(msg));
uint64_t path = Endian_bigEndianToHost64(node.path_be);
uint32_t protocolVersion = Endian_bigEndianToHost32(node.version_be);
const char* log = NULL;
if (!ch->activeSnode.path) {
if (node.path_be) {
log = "Found snode";
} else {
// didn't know the snode before, still don't
return NULL;
}
} else if (!node.path_be) {
// We had one, now we don't
log = "Removing snode";
} else {
if (Bits_memcmp(ch->activeSnode.key, node.publicKey, 32)) {
log = "Changing snode";
} else if (path != ch->activeSnode.path) {
log = "Changing snode path";
} else if (ch->activeSnode.protocolVersion != protocolVersion) {
log = "Changing snode protocolVersion";
} else {
// Nothing has changed
return NULL;
}
}
struct Address old = {
.protocolVersion = ch->activeSnode.protocolVersion,
.path = ch->activeSnode.path,
};
Bits_memcpy(old.key, ch->activeSnode.key, 32);
Bits_memcpy(ch->activeSnode.key, node.publicKey, 32);
ch->activeSnode.path = path;
ch->activeSnode.protocolVersion = protocolVersion;
struct Address addr = { .protocolVersion = protocolVersion, .path = path, };
Bits_memcpy(addr.key, node.publicKey, 32);
Log_debug(ch->log, "%s [%s] -> [%s]", log,
Address_toStringKey(&old, Message_getAlloc(msg))->bytes,
Address_toStringKey(&addr, Message_getAlloc(msg))->bytes);
return NULL;
}
struct ControlHandler* ControlHandler_new(struct Allocator* allocator,
struct Log* logger,
struct EventEmitter* ee,
uint8_t myPublicKey[32])
{
struct Allocator* alloc = Allocator_child(allocator);
struct ControlHandler_pvt* ch = Allocator_calloc(alloc, sizeof(struct ControlHandler_pvt), 1);
ch->alloc = alloc;
ch->log = logger;
Bits_memcpy(ch->myPublicKey, myPublicKey, 32);
ch->pub.coreIf.send = incomingFromCore;
ch->pub.switchPingerIf.send = incomingFromSwitchPinger;
ch->eventIf.send = changeSnode;
EventEmitter_regCore(ee, &ch->eventIf, PFChan_Pathfinder_SNODE);
Identity_set(ch);
return &ch->pub;
}