ControlHandler.c 15 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402
  1. /* vim: set expandtab ts=4 sw=4: */
  2. /*
  3. * You may redistribute this program and/or modify it under the terms of
  4. * the GNU General Public License as published by the Free Software Foundation,
  5. * either version 3 of the License, or (at your option) any later version.
  6. *
  7. * This program is distributed in the hope that it will be useful,
  8. * but WITHOUT ANY WARRANTY; without even the implied warranty of
  9. * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  10. * GNU General Public License for more details.
  11. *
  12. * You should have received a copy of the GNU General Public License
  13. * along with this program. If not, see <https://www.gnu.org/licenses/>.
  14. */
  15. #include "dht/Address.h"
  16. #include "net/ControlHandler.h"
  17. #include "util/Identity.h"
  18. #include "util/AddrTools.h"
  19. #include "util/Checksum.h"
  20. #include "wire/Control.h"
  21. #include "wire/Error.h"
  22. #define NumberCompress_OLD_CODE
  23. #include "switch/NumberCompress.h"
  24. #include <inttypes.h>
  25. struct ControlHandler_pvt
  26. {
  27. struct ControlHandler pub;
  28. struct Log* log;
  29. struct Allocator* alloc;
  30. uint8_t myPublicKey[32];
  31. struct Iface eventIf;
  32. struct Address activeSnode;
  33. Identity
  34. };
  35. /**
  36. * Expects [ Ctrl ][ Error ][ cause SwitchHeader ][ cause handle ][ cause etc.... ]
  37. */
  38. #define handleError_MIN_SIZE (Control_Header_SIZE + Control_Error_MIN_SIZE + SwitchHeader_SIZE + 4)
  39. static Iface_DEFUN handleError(struct Message* msg,
  40. struct ControlHandler_pvt* ch,
  41. uint64_t label,
  42. uint8_t* labelStr,
  43. struct RouteHeader* rh)
  44. {
  45. if (Message_getLength(msg) < handleError_MIN_SIZE) {
  46. Log_info(ch->log, "DROP runt error packet from [%s]", labelStr);
  47. return Error(msg, "RUNT");
  48. }
  49. Er_assert(Message_truncate(msg, handleError_MIN_SIZE));
  50. Er_assert(Message_epush(msg, &rh->sh, SwitchHeader_SIZE));
  51. Er_assert(Message_epush32be(msg, 0xffffffff));
  52. Er_assert(Message_epush32be(msg, PFChan_Core_SWITCH_ERR));
  53. return Iface_next(&ch->eventIf, msg);
  54. }
  55. /**
  56. * Expects [ SwitchHeader ][ Ctrl ][ (key)Ping ][ data etc.... ]
  57. */
  58. #define handlePing_MIN_SIZE (Control_Header_SIZE + Control_Ping_MIN_SIZE)
  59. static Iface_DEFUN handlePing(struct Message* msg,
  60. struct ControlHandler_pvt* ch,
  61. uint64_t label,
  62. uint8_t* labelStr,
  63. uint16_t messageType_be)
  64. {
  65. if (Message_getLength(msg) < handlePing_MIN_SIZE) {
  66. Log_info(ch->log, "DROP runt ping");
  67. return Error(msg, "RUNT");
  68. }
  69. struct Control* ctrl = (struct Control*) msg->msgbytes;
  70. Er_assert(Message_eshift(msg, -Control_Header_SIZE));
  71. // Ping and keyPing share version location
  72. struct Control_Ping* ping = (struct Control_Ping*) msg->msgbytes;
  73. uint32_t herVersion = Endian_bigEndianToHost32(ping->version_be);
  74. if (!Version_isCompatible(Version_CURRENT_PROTOCOL, herVersion)) {
  75. Log_debug(ch->log, "DROP ping from incompatible version [%d]", herVersion);
  76. return Error(msg, "INVALID");
  77. }
  78. if (messageType_be == Control_KEYPING_be) {
  79. Log_debug(ch->log, "got switch keyPing from [%s]", labelStr);
  80. if (Message_getLength(msg) < Control_KeyPing_HEADER_SIZE) {
  81. // min keyPing size is longer
  82. Log_debug(ch->log, "DROP runt keyPing");
  83. return Error(msg, "RUNT");
  84. }
  85. if (Message_getLength(msg) > Control_KeyPing_MAX_SIZE) {
  86. Log_debug(ch->log, "DROP long keyPing");
  87. return Error(msg, "INVALID");
  88. }
  89. if (ping->magic != Control_KeyPing_MAGIC) {
  90. Log_debug(ch->log, "DROP keyPing (bad magic)");
  91. return Error(msg, "INVALID");
  92. }
  93. struct Control_KeyPing* keyPing = (struct Control_KeyPing*) msg->msgbytes;
  94. keyPing->magic = Control_KeyPong_MAGIC;
  95. ctrl->header.type_be = Control_KEYPONG_be;
  96. Bits_memcpy(keyPing->key, ch->myPublicKey, 32);
  97. } else if (messageType_be == Control_PING_be) {
  98. // Happens in benchmark.
  99. //Log_debug(ch->log, "got switch ping from [%s]", labelStr);
  100. if (ping->magic != Control_Ping_MAGIC) {
  101. Log_debug(ch->log, "DROP ping (bad magic)");
  102. return Error(msg, "INVALID");
  103. }
  104. ping->magic = Control_Pong_MAGIC;
  105. ctrl->header.type_be = Control_PONG_be;
  106. } else {
  107. Assert_failure("2+2=5");
  108. }
  109. ping->version_be = Endian_hostToBigEndian32(Version_CURRENT_PROTOCOL);
  110. Er_assert(Message_eshift(msg, Control_Header_SIZE));
  111. ctrl->header.checksum_be = 0;
  112. ctrl->header.checksum_be = Checksum_engine_be(msg->msgbytes, Message_getLength(msg));
  113. Er_assert(Message_eshift(msg, RouteHeader_SIZE));
  114. struct RouteHeader* routeHeader = (struct RouteHeader*) msg->msgbytes;
  115. Bits_memset(routeHeader, 0, RouteHeader_SIZE);
  116. SwitchHeader_setVersion(&routeHeader->sh, SwitchHeader_CURRENT_VERSION);
  117. routeHeader->sh.label_be = Endian_hostToBigEndian64(label);
  118. routeHeader->flags |= RouteHeader_flags_CTRLMSG;
  119. return Iface_next(&ch->pub.coreIf, msg);
  120. }
  121. /**
  122. * Expects [ SwitchHeader ][ Ctrl ][ RPath ][ data etc.... ]
  123. */
  124. #define handleRPathQuery_MIN_SIZE (Control_Header_SIZE + Control_RPath_HEADER_SIZE)
  125. static Iface_DEFUN handleRPathQuery(struct Message* msg,
  126. struct ControlHandler_pvt* ch,
  127. uint64_t label,
  128. uint8_t* labelStr)
  129. {
  130. Log_debug(ch->log, "Incoming RPATH query");
  131. if (Message_getLength(msg) < handleRPathQuery_MIN_SIZE) {
  132. Log_info(ch->log, "DROP runt RPATH query");
  133. return Error(msg, "RUNT");
  134. }
  135. struct Control* ctrl = (struct Control*) msg->msgbytes;
  136. struct Control_RPath* rpa = &ctrl->content.rpath;
  137. if (rpa->magic != Control_RPATH_QUERY_MAGIC) {
  138. Log_debug(ch->log, "DROP RPATH query (bad magic)");
  139. return Error(msg, "INVALID");
  140. }
  141. ctrl->header.type_be = Control_RPATH_REPLY_be;
  142. rpa->version_be = Endian_hostToBigEndian32(Version_CURRENT_PROTOCOL);
  143. rpa->magic = Control_RPATH_REPLY_MAGIC;
  144. uint64_t label_be = Endian_hostToBigEndian64(label);
  145. Bits_memcpy(rpa->rpath_be, &label_be, 8);
  146. ctrl->header.checksum_be = 0;
  147. ctrl->header.checksum_be = Checksum_engine_be(msg->msgbytes, Message_getLength(msg));
  148. Er_assert(Message_eshift(msg, RouteHeader_SIZE));
  149. struct RouteHeader* routeHeader = (struct RouteHeader*) msg->msgbytes;
  150. Bits_memset(routeHeader, 0, RouteHeader_SIZE);
  151. SwitchHeader_setVersion(&routeHeader->sh, SwitchHeader_CURRENT_VERSION);
  152. routeHeader->sh.label_be = label_be;
  153. routeHeader->flags |= RouteHeader_flags_CTRLMSG;
  154. return Iface_next(&ch->pub.coreIf, msg);
  155. }
  156. /**
  157. * Expects [ SwitchHeader ][ Ctrl ][ SupernodeQuery ][ data etc.... ]
  158. */
  159. #define handleGetSnodeQuery_MIN_SIZE (Control_Header_SIZE + Control_GetSnode_HEADER_SIZE)
  160. static Iface_DEFUN handleGetSnodeQuery(struct Message* msg,
  161. struct ControlHandler_pvt* ch,
  162. uint64_t label,
  163. uint8_t* labelStr)
  164. {
  165. Log_debug(ch->log, "incoming getSupernode query");
  166. if (Message_getLength(msg) < handleGetSnodeQuery_MIN_SIZE) {
  167. Log_info(ch->log, "DROP runt getSupernode query");
  168. return Error(msg, "RUNT");
  169. }
  170. struct Control* ctrl = (struct Control*) msg->msgbytes;
  171. struct Control_GetSnode* snq = &ctrl->content.getSnode;
  172. if (snq->magic != Control_GETSNODE_QUERY_MAGIC) {
  173. Log_debug(ch->log, "DROP getSupernode query (bad magic)");
  174. return Error(msg, "INVALID");
  175. }
  176. uint32_t herVersion = Endian_bigEndianToHost32(snq->version_be);
  177. if (!Version_isCompatible(Version_CURRENT_PROTOCOL, herVersion)) {
  178. Log_debug(ch->log, "DROP getSupernode query from incompatible version [%d]", herVersion);
  179. // Nothing wrong with the query but we're just not going to answer it
  180. return NULL;
  181. }
  182. ctrl->header.type_be = Control_GETSNODE_REPLY_be;
  183. snq->kbps_be = 0xffffffff;
  184. snq->version_be = Endian_hostToBigEndian32(Version_CURRENT_PROTOCOL);
  185. snq->magic = Control_GETSNODE_REPLY_MAGIC;
  186. uint64_t fixedLabel = 0;
  187. if (ch->activeSnode.path) {
  188. fixedLabel = NumberCompress_getLabelFor(ch->activeSnode.path, label);
  189. uint64_t fixedLabel_be = Endian_hostToBigEndian64(fixedLabel);
  190. Bits_memcpy(snq->pathToSnode_be, &fixedLabel_be, 8);
  191. Bits_memcpy(snq->snodeKey, ch->activeSnode.key, 32);
  192. snq->snodeVersion_be = Endian_hostToBigEndian32(ch->activeSnode.protocolVersion);
  193. } else {
  194. Log_debug(ch->log, "getSupernode query and we do not have a known snode");
  195. snq->snodeVersion_be = 0;
  196. Bits_memset(snq->pathToSnode_be, 0, 8);
  197. Bits_memset(snq->snodeKey, 0, 32);
  198. }
  199. ctrl->header.checksum_be = 0;
  200. ctrl->header.checksum_be = Checksum_engine_be(msg->msgbytes, Message_getLength(msg));
  201. Er_assert(Message_eshift(msg, RouteHeader_SIZE));
  202. struct RouteHeader* routeHeader = (struct RouteHeader*) msg->msgbytes;
  203. Bits_memset(routeHeader, 0, RouteHeader_SIZE);
  204. SwitchHeader_setVersion(&routeHeader->sh, SwitchHeader_CURRENT_VERSION);
  205. routeHeader->sh.label_be = Endian_hostToBigEndian64(label);
  206. routeHeader->flags |= RouteHeader_flags_CTRLMSG;
  207. Log_debug(ch->log, "Replied to [%" PRIx64 "] with snode [%" PRIx64 "] aka [%" PRIx64 "]",
  208. label, fixedLabel, ch->activeSnode.path);
  209. return Iface_next(&ch->pub.coreIf, msg);
  210. }
  211. /**
  212. * Handle an incoming control message from a switch.
  213. *
  214. * @param context the ducttape context.
  215. * @param message the control message, this should be alligned on the beginning of the content,
  216. * that is to say, after the end of the switch header.
  217. * @param switchHeader the header.
  218. * @param switchIf the interface which leads to the switch.
  219. * @param isFormV8 true if the control message is in the form specified by protocol version 8+
  220. */
  221. static Iface_DEFUN incomingFromCore(struct Message* msg, struct Iface* coreIf)
  222. {
  223. struct ControlHandler_pvt* ch = Identity_check((struct ControlHandler_pvt*) coreIf);
  224. struct RouteHeader routeHdr;
  225. Er_assert(Message_epop(msg, &routeHdr, RouteHeader_SIZE));
  226. uint8_t labelStr[20];
  227. uint64_t label = Endian_bigEndianToHost64(routeHdr.sh.label_be);
  228. AddrTools_printPath(labelStr, label);
  229. // happens in benchmark
  230. // Log_debug(ch->log, "ctrl packet from [%s]", labelStr);
  231. if (Message_getLength(msg) < 4 + Control_Header_SIZE) {
  232. Log_info(ch->log, "DROP runt ctrl packet from [%s]", labelStr);
  233. return Error(msg, "RUNT");
  234. }
  235. Assert_true(routeHdr.flags & RouteHeader_flags_CTRLMSG);
  236. if (Checksum_engine_be(msg->msgbytes, Message_getLength(msg))) {
  237. Log_info(ch->log, "DROP ctrl packet from [%s] with invalid checksum", labelStr);
  238. return Error(msg, "INVALID");
  239. }
  240. struct Control* ctrl = (struct Control*) msg->msgbytes;
  241. if (ctrl->header.type_be == Control_ERROR_be) {
  242. return handleError(msg, ch, label, labelStr, &routeHdr);
  243. } else if (ctrl->header.type_be == Control_KEYPING_be
  244. || ctrl->header.type_be == Control_PING_be)
  245. {
  246. return handlePing(msg, ch, label, labelStr, ctrl->header.type_be);
  247. } else if (ctrl->header.type_be == Control_KEYPONG_be
  248. || ctrl->header.type_be == Control_PONG_be)
  249. {
  250. Log_debug(ch->log, "got switch pong from [%s]", labelStr);
  251. Er_assert(Message_epush(msg, &routeHdr, RouteHeader_SIZE));
  252. return Iface_next(&ch->pub.switchPingerIf, msg);
  253. } else if (ctrl->header.type_be == Control_GETSNODE_QUERY_be) {
  254. return handleGetSnodeQuery(msg, ch, label, labelStr);
  255. } else if (ctrl->header.type_be == Control_GETSNODE_REPLY_be
  256. || ctrl->header.type_be == Control_RPATH_REPLY_be)
  257. {
  258. Log_debug(ch->log, "got %s REPLY from [%s]",
  259. (ctrl->header.type_be == Control_GETSNODE_REPLY_be) ? "GETSNODE" : "RPATH",
  260. labelStr);
  261. Er_assert(Message_epush(msg, &routeHdr, RouteHeader_SIZE));
  262. Er_assert(Message_epush32be(msg, 0xffffffff));
  263. Er_assert(Message_epush32be(msg, PFChan_Core_CTRL_MSG));
  264. return Iface_next(&ch->eventIf, msg);
  265. } else if (ctrl->header.type_be == Control_RPATH_QUERY_be) {
  266. return handleRPathQuery(msg, ch, label, labelStr);
  267. }
  268. Log_info(ch->log, "DROP control packet of unknown type from [%s], type [%d]",
  269. labelStr, Endian_bigEndianToHost16(ctrl->header.type_be));
  270. return Error(msg, "INVALID");
  271. }
  272. // Forward from switch pinger directly to core.
  273. static Iface_DEFUN incomingFromSwitchPinger(struct Message* msg, struct Iface* switchPingerIf)
  274. {
  275. struct ControlHandler_pvt* ch =
  276. Identity_containerOf(switchPingerIf, struct ControlHandler_pvt, pub.switchPingerIf);
  277. return Iface_next(&ch->pub.coreIf, msg);
  278. }
  279. static Iface_DEFUN changeSnode(struct Message* msg, struct Iface* eventIf)
  280. {
  281. struct ControlHandler_pvt* ch =
  282. Identity_containerOf(eventIf, struct ControlHandler_pvt, eventIf);
  283. enum PFChan_Pathfinder ev = Er_assert(Message_epop32be(msg));
  284. Assert_true(ev == PFChan_Pathfinder_SNODE);
  285. Er_assert(Message_epop32be(msg));
  286. struct PFChan_Node node;
  287. Er_assert(Message_epop(msg, &node, PFChan_Node_SIZE));
  288. Assert_true(!Message_getLength(msg));
  289. uint64_t path = Endian_bigEndianToHost64(node.path_be);
  290. uint32_t protocolVersion = Endian_bigEndianToHost32(node.version_be);
  291. const char* log = NULL;
  292. if (!ch->activeSnode.path) {
  293. if (node.path_be) {
  294. log = "Found snode";
  295. } else {
  296. // didn't know the snode before, still don't
  297. return NULL;
  298. }
  299. } else if (!node.path_be) {
  300. // We had one, now we don't
  301. log = "Removing snode";
  302. } else {
  303. if (Bits_memcmp(ch->activeSnode.key, node.publicKey, 32)) {
  304. log = "Changing snode";
  305. } else if (path != ch->activeSnode.path) {
  306. log = "Changing snode path";
  307. } else if (ch->activeSnode.protocolVersion != protocolVersion) {
  308. log = "Changing snode protocolVersion";
  309. } else {
  310. // Nothing has changed
  311. return NULL;
  312. }
  313. }
  314. struct Address old = {
  315. .protocolVersion = ch->activeSnode.protocolVersion,
  316. .path = ch->activeSnode.path,
  317. };
  318. Bits_memcpy(old.key, ch->activeSnode.key, 32);
  319. Bits_memcpy(ch->activeSnode.key, node.publicKey, 32);
  320. ch->activeSnode.path = path;
  321. ch->activeSnode.protocolVersion = protocolVersion;
  322. struct Address addr = { .protocolVersion = protocolVersion, .path = path, };
  323. Bits_memcpy(addr.key, node.publicKey, 32);
  324. Log_debug(ch->log, "%s [%s] -> [%s]", log,
  325. Address_toStringKey(&old, Message_getAlloc(msg))->bytes,
  326. Address_toStringKey(&addr, Message_getAlloc(msg))->bytes);
  327. return NULL;
  328. }
  329. struct ControlHandler* ControlHandler_new(struct Allocator* allocator,
  330. struct Log* logger,
  331. struct EventEmitter* ee,
  332. uint8_t myPublicKey[32])
  333. {
  334. struct Allocator* alloc = Allocator_child(allocator);
  335. struct ControlHandler_pvt* ch = Allocator_calloc(alloc, sizeof(struct ControlHandler_pvt), 1);
  336. ch->alloc = alloc;
  337. ch->log = logger;
  338. Bits_memcpy(ch->myPublicKey, myPublicKey, 32);
  339. ch->pub.coreIf.send = incomingFromCore;
  340. ch->pub.switchPingerIf.send = incomingFromSwitchPinger;
  341. ch->eventIf.send = changeSnode;
  342. EventEmitter_regCore(ee, &ch->eventIf, PFChan_Pathfinder_SNODE);
  343. Identity_set(ch);
  344. return &ch->pub;
  345. }