/* 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 "exception/Err.h"
#include "memory/Allocator.h"
#include "net/InterfaceController.h"
#include "switch/EncodingScheme.h"
#include "util/Bits.h"
#include "util/Endian.h"
#include "util/Identity.h"
#include "util/AddrTools.h"
#include "util/Checksum.h"
#include "util/platform/Sockaddr.h"
#include "util/version/Version.h"
#include "wire/Control.h"
#include "wire/Error.h"
#include "wire/Message.h"
#define NumberCompress_OLD_CODE
#include "switch/NumberCompress.h"
#include
struct ControlHandler_pvt
{
struct ControlHandler pub;
struct EncodingScheme* ourEncodingScheme;
struct Log* log;
struct Allocator* alloc;
struct InterfaceController* ifc;
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(Message_t* 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");
}
Err(Message_truncate(msg, handleError_MIN_SIZE));
Err(Message_epush(msg, &rh->sh, SwitchHeader_SIZE));
Err(Message_epush32be(msg, 0xffffffff));
Err(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(Message_t* 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*) Message_bytes(msg);
Err(Message_eshift(msg, -Control_Header_SIZE));
// Ping and keyPing share version location
struct Control_Ping* ping = (struct Control_Ping*) Message_bytes(msg);
uint32_t herVersion = Endian_bigEndianToHost32(ping->version_be);
if (!Version_compatibleWithCurrent(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*) Message_bytes(msg);
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);
Err(Message_eshift(msg, Control_Header_SIZE));
ctrl->header.checksum_be = 0;
ctrl->header.checksum_be = Checksum_engine_be(Message_bytes(msg), Message_getLength(msg));
Err(Message_eshift(msg, RouteHeader_SIZE));
struct RouteHeader* routeHeader = (struct RouteHeader*) Message_bytes(msg);
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(Message_t* 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*) Message_bytes(msg);
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(Message_bytes(msg), Message_getLength(msg));
Err(Message_eshift(msg, RouteHeader_SIZE));
struct RouteHeader* routeHeader = (struct RouteHeader*) Message_bytes(msg);
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);
}
static Err_DEFUN writeLlAddr(Message_t* msg, Sockaddr_t* sa) {
struct Control_LlAddr out = {
.magic = Control_LlAddr_REPLY_MAGIC,
.version_be = Endian_hostToBigEndian32(Version_CURRENT_PROTOCOL),
};
if (sa->type == Sockaddr_PLATFORM) {
int fam = Sockaddr_getFamily(sa);
int port = Sockaddr_getPort(sa);
if (fam == Sockaddr_AF_INET && port > -1) {
out.addr.udp4.type = Control_LlAddr_Udp4_TYPE;
out.addr.udp4.len = sizeof(Control_LlAddr_Udp4_t);
out.addr.udp4.port_be = Endian_hostToBigEndian16(port);
uint8_t* p = NULL;
int len = Sockaddr_getAddress(sa, &p);
Assert_true(len == sizeof out.addr.udp4.addr && p != NULL);
Bits_memcpy(&out.addr.udp4.addr, p, sizeof out.addr.udp4.addr);
return Message_epush(msg, &out, sizeof out);
} else if (fam == Sockaddr_AF_INET6 && port > -1) {
out.addr.udp6.type = Control_LlAddr_Udp6_TYPE;
out.addr.udp6.len = sizeof(Control_LlAddr_Udp6_t);
out.addr.udp6.port_be = Endian_hostToBigEndian16(port);
uint8_t* p = NULL;
int len = Sockaddr_getAddress(sa, &p);
Assert_true(len == sizeof out.addr.udp6.addr && p != NULL);
Bits_memcpy(&out.addr.udp6.addr, p, sizeof out.addr.udp6.addr);
return Message_epush(msg, &out, sizeof out);
}
}
out.addr.other.type = Control_LlAddr_Other_TYPE;
out.addr.other.len = sizeof(Control_LlAddr_Other_t);
unsigned int len = sa->addrLen;
if (len > sizeof out.addr.other.sockaddrHeader) {
len = sizeof out.addr.other.sockaddrHeader;
}
Bits_memcpy(&out.addr.other.sockaddrHeader, &sa, len);
return Message_epush(msg, &out, sizeof out);
}
/**
* Expects [ SwitchHeader ][ Ctrl ][ LlAddr ]
*/
#define handleLlAddrQuery_MIN_SIZE (Control_Header_SIZE + Control_LlAddr_HEADER_SIZE)
static Iface_DEFUN handleLlAddrQuery(Message_t* msg,
struct ControlHandler_pvt* ch,
uint64_t label,
uint8_t* labelStr)
{
Log_debug(ch->log, "Incoming LLADDR query");
if (Message_getLength(msg) < handleLlAddrQuery_MIN_SIZE) {
Log_info(ch->log, "DROP runt LLADDR query");
return Error(msg, "RUNT");
}
struct Control_Header hdr;
Err(Message_epop(msg, &hdr, sizeof hdr));
struct Control_LlAddr lla;
Err(Message_epop(msg, &lla, sizeof lla));
if (lla.magic != Control_LlAddr_QUERY_MAGIC) {
Log_debug(ch->log, "DROP LLADDR query (bad magic)");
return Error(msg, "INVALID");
}
// If this is not a one-hop packet, it's invalid
if (!EncodingScheme_isOneHop(ch->ourEncodingScheme, label)) {
Log_debug(ch->log, "DROP LLADDR query from non-peer");
return Error(msg, "INVALID");
}
// Get the lladdr from the peer who sent the message
struct Sockaddr* sa = InterfaceController_getPeerLlAddr(
ch->ifc, Message_getAlloc(msg), label);
if (!sa) {
Log_info(ch->log, "LLADDR query peer not found");
return Error(msg, "INTERNAL");
}
Err(writeLlAddr(msg, sa));
hdr.type_be = Control_LlAddr_REPLY_be;
hdr.checksum_be = 0;
Err(Message_epush(msg, &hdr, sizeof hdr));
hdr.checksum_be = Checksum_engine_be(Message_bytes(msg), Message_getLength(msg));
Err(Message_epop(msg, NULL, sizeof hdr));
Err(Message_epush(msg, &hdr, sizeof hdr));
struct RouteHeader routeHeader = {
.sh.label_be = Endian_hostToBigEndian64(label),
.flags = RouteHeader_flags_CTRLMSG,
};
SwitchHeader_setVersion(&routeHeader.sh, SwitchHeader_CURRENT_VERSION);
Err(Message_epush(msg, &routeHeader, sizeof routeHeader));
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(Message_t* 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*) Message_bytes(msg);
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_compatibleWithCurrent(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(Message_bytes(msg), Message_getLength(msg));
Err(Message_eshift(msg, RouteHeader_SIZE));
struct RouteHeader* routeHeader = (struct RouteHeader*) Message_bytes(msg);
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(Message_t* msg, struct Iface* coreIf)
{
struct ControlHandler_pvt* ch = Identity_check((struct ControlHandler_pvt*) coreIf);
struct RouteHeader routeHdr;
Err(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(Message_bytes(msg), 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*) Message_bytes(msg);
const uint16_t t = ctrl->header.type_be;
if (t == Control_ERROR_be) {
return handleError(msg, ch, label, labelStr, &routeHdr);
} else if (t == Control_KEYPING_be || t == Control_PING_be) {
return handlePing(msg, ch, label, labelStr, t);
} else if (t == Control_GETSNODE_QUERY_be) {
return handleGetSnodeQuery(msg, ch, label, labelStr);
} else if (t == Control_RPATH_QUERY_be) {
return handleRPathQuery(msg, ch, label, labelStr);
} else if (t == Control_LlAddr_QUERY_be) {
return handleLlAddrQuery(msg, ch, label, labelStr);
} else if (
t == Control_PONG_be ||
t == Control_KEYPONG_be ||
t == Control_LlAddr_REPLY_be ||
t == Control_GETSNODE_REPLY_be ||
t == Control_RPATH_REPLY_be)
{
if (t == Control_PONG_be) {
// Special case for IfController which has it's own pinger
int32_t offset = Message_getLength(msg) - 12;
uint8_t* ptr = &Message_bytes(msg)[offset];
if (!Bits_memcmp(ptr, "IFACE_CNTRLR", 12)) {
Log_debug(ch->log, "IFC switch pong from [%s]", labelStr);
Err(Message_epush(msg, &routeHdr, RouteHeader_SIZE));
return Iface_next(&ch->pub.ifcSwitchPingerIf, msg);
}
}
Log_debug(ch->log, "Got [%d] REPLY from [%s]", t, labelStr);
Err(Message_epush(msg, &routeHdr, RouteHeader_SIZE));
Err(Message_epush32be(msg, 0xffffffff));
Err(Message_epush32be(msg, PFChan_Core_CTRL_MSG));
return Iface_next(&ch->eventIf, msg);
}
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(Message_t* msg, struct Iface* switchPingerIf)
{
struct ControlHandler_pvt* ch =
Identity_containerOf(switchPingerIf, struct ControlHandler_pvt, pub.ifcSwitchPingerIf);
return Iface_next(&ch->pub.coreIf, msg);
}
static Iface_DEFUN changeSnode(Message_t* msg, struct Iface* eventIf)
{
struct ControlHandler_pvt* ch =
Identity_containerOf(eventIf, struct ControlHandler_pvt, eventIf);
enum PFChan_Pathfinder ev = 0;
Err(Message_epop32be(&ev, msg));
Assert_true(ev == PFChan_Pathfinder_SNODE);
uint32_t discard = 0;
Err(Message_epop32be(&discard, msg));
struct PFChan_Node node;
Err(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 InterfaceController* ifc)
{
struct Allocator* alloc = Allocator_child(allocator);
struct ControlHandler_pvt* ch = Allocator_calloc(alloc, sizeof(struct ControlHandler_pvt), 1);
ch->ourEncodingScheme = NumberCompress_defineScheme(alloc);
ch->alloc = alloc;
ch->log = logger;
ch->ifc = ifc;
Bits_memcpy(ch->myPublicKey, myPublicKey, 32);
ch->pub.coreIf.send = incomingFromCore;
ch->pub.ifcSwitchPingerIf.send = incomingFromSwitchPinger;
ch->eventIf.send = changeSnode;
EventEmitter_regCore(ee, &ch->eventIf, PFChan_Pathfinder_SNODE);
Identity_set(ch);
return &ch->pub;
}