SubnodePathfinder.c 23 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644
  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 "subnode/SubnodePathfinder.h"
  16. #include "benc/Object.h"
  17. #include "interface/Iface.h"
  18. #include "rust/cjdns_sys/RTypes.h"
  19. #include "rust/cjdns_sys/Rffi.h"
  20. #include "subnode/AddrSet.h"
  21. #include "subnode/MsgCore.h"
  22. #include "subnode/PeeringSeeder.h"
  23. #include "subnode/SupernodeHunter.h"
  24. #include "subnode/PingResponder.h"
  25. #include "subnode/BoilerplateResponder.h"
  26. #include "subnode/ReachabilityCollector.h"
  27. #include "crypto/AddressCalc.h"
  28. #include "dht/Address.h"
  29. #include "util/log/Log.h"
  30. #include "wire/DataHeader.h"
  31. #include "wire/RouteHeader.h"
  32. #include "dht/dhtcore/ReplySerializer.h"
  33. #include "util/AddrTools.h"
  34. #include "util/events/Timeout.h"
  35. #include "net/SwitchPinger.h"
  36. #include "switch/LabelSplicer.h"
  37. #include "wire/Error.h"
  38. #include "wire/PFChan.h"
  39. #include "wire/DataHeader.h"
  40. #include "util/CString.h"
  41. #include "wire/Metric.h"
  42. #include "util/Gcc.h"
  43. #include "subnode/ReachabilityAnnouncer.h"
  44. struct Query {
  45. struct Address target;
  46. uint8_t routeFrom[16];
  47. uint8_t routeTo[16];
  48. };
  49. #define Map_NAME OfPromiseByQuery
  50. #define Map_KEY_TYPE struct Query
  51. #define Map_VALUE_TYPE struct MsgCore_Promise*
  52. #define Map_ENABLE_HANDLES
  53. #include "util/Map.h"
  54. struct SubnodePathfinder_pvt
  55. {
  56. struct SubnodePathfinder pub;
  57. struct Iface msgCoreIf;
  58. struct Allocator* alloc;
  59. struct Log* log;
  60. EventBase_t* base;
  61. struct Random* rand;
  62. #define SubnodePathfinder_pvt_state_INITIALIZING 0
  63. #define SubnodePathfinder_pvt_state_RUNNING 1
  64. int state;
  65. struct Address* myAddress;
  66. struct AddrSet* myPeerAddrs;
  67. struct MsgCore* msgCore;
  68. struct BoilerplateResponder* br;
  69. struct ReachabilityAnnouncer* ra;
  70. struct Map_OfPromiseByQuery queryMap;
  71. struct Iface switchPingerIf;
  72. struct EncodingScheme* myScheme;
  73. uint8_t* privateKey;
  74. String* encodingSchemeStr;
  75. Iface_t seederIf;
  76. Ca_t* ca;
  77. Identity
  78. };
  79. static void nodeForAddress(struct PFChan_Node* nodeOut, struct Address* addr, uint32_t metric)
  80. {
  81. Bits_memset(nodeOut, 0, PFChan_Node_SIZE);
  82. nodeOut->version_be = Endian_hostToBigEndian32(addr->protocolVersion);
  83. nodeOut->metric_be = Endian_hostToBigEndian32(metric);
  84. nodeOut->path_be = Endian_hostToBigEndian64(addr->path);
  85. Bits_memcpy(nodeOut->publicKey, addr->key, 32);
  86. Bits_memcpy(nodeOut->ip6, addr->ip6.bytes, 16);
  87. }
  88. static Iface_DEFUN sendNode(Message_t* msg,
  89. struct Address* addr,
  90. uint32_t metric,
  91. enum PFChan_Pathfinder msgType,
  92. struct SubnodePathfinder_pvt* pf)
  93. {
  94. Message_reset(msg);
  95. Err(Message_eshift(msg, PFChan_Node_SIZE));
  96. nodeForAddress((struct PFChan_Node*) Message_bytes(msg), addr, metric);
  97. if (addr->path == UINT64_MAX) {
  98. ((struct PFChan_Node*) Message_bytes(msg))->path_be = 0;
  99. }
  100. Err(Message_epush32be(msg, msgType));
  101. return Iface_next(&pf->pub.eventIf, msg);
  102. }
  103. static Iface_DEFUN connected(struct SubnodePathfinder_pvt* pf, Gcc_UNUSED Message_t* msg)
  104. {
  105. Log_debug(pf->log, "INIT");
  106. pf->state = SubnodePathfinder_pvt_state_RUNNING;
  107. return NULL;
  108. }
  109. static uint32_t addressForNode(struct Address* addrOut, Message_t* msg)
  110. {
  111. Bits_memset(addrOut, 0, sizeof(struct Address));
  112. struct PFChan_Node node;
  113. Err_assert(Message_epop(msg, &node, PFChan_Node_SIZE));
  114. Assert_true(!Message_getLength(msg));
  115. addrOut->protocolVersion = Endian_bigEndianToHost32(node.version_be);
  116. addrOut->path = Endian_bigEndianToHost64(node.path_be);
  117. Bits_memcpy(addrOut->key, node.publicKey, 32);
  118. Bits_memcpy(addrOut->ip6.bytes, node.ip6, 16);
  119. return Endian_bigEndianToHost32(node.metric_be);
  120. }
  121. static Iface_DEFUN switchErr(Message_t* msg, struct SubnodePathfinder_pvt* pf)
  122. {
  123. struct PFChan_Core_SwitchErr switchErr;
  124. Err(Message_epop(msg, &switchErr, PFChan_Core_SwitchErr_MIN_SIZE));
  125. uint64_t path = Endian_bigEndianToHost64(switchErr.sh.label_be);
  126. if (pf->pub.snh->snodeAddr.path &&
  127. // pf->pub.snh->snodeAddr.path != path && // ctrl errors from the snode should be same
  128. LabelSplicer_routesThrough(pf->pub.snh->snodeAddr.path, path)) {
  129. uint8_t pathStr[20];
  130. AddrTools_printPath(pathStr, path);
  131. int err = Endian_bigEndianToHost32(switchErr.ctrlErr.errorType_be);
  132. Log_debug(pf->log, "switch err from active snode [%s] type [%d]",
  133. pathStr, err);
  134. pf->pub.snh->snodeIsReachable = false;
  135. if (pf->pub.snh->onSnodeUnreachable) {
  136. pf->pub.snh->onSnodeUnreachable(pf->pub.snh, 0, 0);
  137. }
  138. }
  139. // TODO(cjd): We should be reporting a bad link to the session manager but
  140. // we only really have the ability to report a node with known IPv6 address
  141. // so we will need to add a new event type to PFChan.
  142. return NULL;
  143. }
  144. struct SnodeQuery {
  145. struct SubnodePathfinder_pvt* pf;
  146. uint32_t mapHandle;
  147. Identity
  148. };
  149. static void getRouteReply(Dict* msg, struct Address* src, struct MsgCore_Promise* prom)
  150. {
  151. struct SnodeQuery* snq = Identity_check((struct SnodeQuery*) prom->userData);
  152. struct SubnodePathfinder_pvt* pf = Identity_check(snq->pf);
  153. int index = Map_OfPromiseByQuery_indexForHandle(snq->mapHandle, &pf->queryMap);
  154. Assert_true(index > -1);
  155. Map_OfPromiseByQuery_remove(index, &pf->queryMap);
  156. if (!src) {
  157. Log_debug(pf->log, "GetRoute timeout");
  158. return;
  159. }
  160. Log_debug(pf->log, "Search reply!");
  161. struct Address_List* al = ReplySerializer_parse(src, msg, pf->log, false, prom->alloc);
  162. if (!al || al->length == 0) { return; }
  163. Log_debug(pf->log, "reply with[%s]", Address_toString(&al->elems[0], prom->alloc)->bytes);
  164. if (al->elems[0].protocolVersion < 20) {
  165. Log_debug(pf->log, "not sending [%s] because version is old",
  166. Address_toString(&al->elems[0], prom->alloc)->bytes);
  167. return;
  168. }
  169. //NodeCache_discoverNode(pf->nc, &al->elems[0]);
  170. Message_t* msgToCore = Message_new(0, 512, prom->alloc);
  171. Iface_CALL(sendNode, msgToCore, &al->elems[0], Metric_SNODE_SAYS, PFChan_Pathfinder_NODE, pf);
  172. }
  173. static void queryRs(struct SubnodePathfinder_pvt* pf, uint8_t addr[16], uint8_t printedAddr[40])
  174. {
  175. if (!pf->pub.snh || !pf->pub.snh->snodeAddr.path) {
  176. Log_debug(pf->log, "Search for [%s] impossible because we have no snode", printedAddr);
  177. return;
  178. }
  179. struct Query q = { .routeFrom = { 0 } };
  180. Bits_memcpy(&q.target, &pf->pub.snh->snodeAddr, sizeof(struct Address));
  181. Bits_memcpy(q.routeFrom, pf->myAddress->ip6.bytes, 16);
  182. Bits_memcpy(q.routeFrom, addr, 16);
  183. if (Map_OfPromiseByQuery_indexForKey(&q, &pf->queryMap) > -1) {
  184. Log_debug(pf->log, "Search for [%s] skipped because one is outstanding", printedAddr);
  185. return;
  186. }
  187. struct MsgCore_Promise* qp = MsgCore_createQuery(pf->msgCore, 0, pf->alloc);
  188. struct SnodeQuery* snq = Allocator_calloc(qp->alloc, sizeof(struct SnodeQuery), 1);
  189. Identity_set(snq);
  190. snq->pf = pf;
  191. Dict* dict = qp->msg = Dict_new(qp->alloc);
  192. qp->cb = getRouteReply;
  193. qp->userData = snq;
  194. Assert_true(AddressCalc_validAddress(pf->pub.snh->snodeAddr.ip6.bytes));
  195. qp->target = &pf->pub.snh->snodeAddr;
  196. Log_debug(pf->log, "Sending getRoute to snode [%s] for [%s]",
  197. Address_toString(qp->target, qp->alloc)->bytes,
  198. printedAddr);
  199. Dict_putStringCC(dict, "sq", "gr", qp->alloc);
  200. String* src = String_newBinary(pf->myAddress->ip6.bytes, 16, qp->alloc);
  201. Dict_putStringC(dict, "src", src, qp->alloc);
  202. String* target = String_newBinary(addr, 16, qp->alloc);
  203. Dict_putStringC(dict, "tar", target, qp->alloc);
  204. int index = Map_OfPromiseByQuery_put(&q, &qp, &pf->queryMap);
  205. snq->mapHandle = pf->queryMap.handles[index];
  206. }
  207. static Iface_DEFUN searchReq(Message_t* msg, struct SubnodePathfinder_pvt* pf)
  208. {
  209. uint8_t addr[16];
  210. Err(Message_epop(msg, addr, 16));
  211. Err(Message_epop(msg, NULL, 4));
  212. uint32_t version = -1;
  213. Err(Message_epop32be(&version, msg));
  214. if (version && version < 20) { return Error(msg, "UNHANDLED"); }
  215. Assert_true(!Message_getLength(msg));
  216. uint8_t printedAddr[40];
  217. AddrTools_printIp(printedAddr, addr);
  218. // We're not going to query for a direct peer because it slows down the process significantly
  219. // and right now, Metric.h always prefers a reported peer to the advice of the snode, which
  220. for (int i = 0; i < pf->myPeerAddrs->length; ++i) {
  221. struct Address* myPeer = AddrSet_get(pf->myPeerAddrs, i);
  222. if (Bits_memcmp(myPeer->ip6.bytes, addr, 16)) { continue; }
  223. Log_debug(pf->log, "Skip for [%s] is our peer, provide that immediately", printedAddr);
  224. // warning: msg is nolonger usable, it has disappeared into sendNode
  225. queryRs(pf, addr, printedAddr);
  226. return sendNode(msg, myPeer, Metric_PF_PEER, PFChan_Pathfinder_NODE, pf);
  227. }
  228. if (pf->pub.snh && !Bits_memcmp(pf->pub.snh->snodeAddr.ip6.bytes, addr, 16)) {
  229. // Querying for a path TO our snode, we can return the path we know right now but also
  230. // make the query...
  231. Log_debug(pf->log, "Skip for [%s] is our snode, provide that immediately", printedAddr);
  232. queryRs(pf, addr, printedAddr);
  233. return sendNode(msg, &pf->pub.snh->snodeAddr, Metric_SNODE, PFChan_Pathfinder_NODE, pf);
  234. }
  235. queryRs(pf, addr, printedAddr);
  236. return NULL;
  237. }
  238. static void rcChange(struct ReachabilityCollector* rc,
  239. struct Address* nodeAddr,
  240. struct ReachabilityCollector_PeerInfo* pi)
  241. {
  242. struct SubnodePathfinder_pvt* pf = Identity_check((struct SubnodePathfinder_pvt*) rc->userData);
  243. ReachabilityAnnouncer_updatePeer(pf->ra, nodeAddr, pi);
  244. }
  245. struct Ping {
  246. struct SubnodePathfinder_pvt* pf;
  247. uint32_t mapHandle;
  248. Identity
  249. };
  250. static void pingReply(Gcc_UNUSED Dict* msg, struct Address* src, struct MsgCore_Promise* prom)
  251. {
  252. struct Ping* usp = Identity_check((struct Ping*) prom->userData);
  253. struct SubnodePathfinder_pvt* pf = Identity_check(usp->pf);
  254. int index = Map_OfPromiseByQuery_indexForHandle(usp->mapHandle, &pf->queryMap);
  255. Assert_true(index > -1);
  256. Map_OfPromiseByQuery_remove(index, &pf->queryMap);
  257. if (!src) {
  258. Log_debug(pf->log, "Ping timeout pinging [%s]",
  259. Address_toString(prom->target, prom->alloc)->bytes);
  260. return;
  261. }
  262. Log_debug(pf->log, "Ping reply from [%s]", Address_toString(src, prom->alloc)->bytes);
  263. Message_t* msgToCore = Message_new(0, 512, prom->alloc);
  264. Iface_CALL(sendNode, msgToCore, src, Metric_PING_REPLY, PFChan_Pathfinder_NODE, pf);
  265. }
  266. static void pingNode(struct SubnodePathfinder_pvt* pf, struct Address* addr)
  267. {
  268. struct Query q = { .routeFrom = { 0 } };
  269. Bits_memcpy(&q.target, addr, sizeof(struct Address));
  270. if (Map_OfPromiseByQuery_indexForKey(&q, &pf->queryMap) > -1) {
  271. Log_debug(pf->log, "Skipping ping because one is already outstanding");
  272. return;
  273. }
  274. // We have a path to the node but the session is not setup, lets ping them...
  275. struct MsgCore_Promise* qp = MsgCore_createQuery(pf->msgCore, 0, pf->alloc);
  276. struct Ping* usp = Allocator_calloc(qp->alloc, sizeof(struct Ping), 1);
  277. Identity_set(usp);
  278. usp->pf = pf;
  279. Dict* dict = qp->msg = Dict_new(qp->alloc);
  280. qp->cb = pingReply;
  281. qp->userData = usp;
  282. Assert_true(AddressCalc_validAddress(addr->ip6.bytes));
  283. Assert_true(addr->path);
  284. qp->target = Address_clone(addr, qp->alloc);
  285. Log_debug(pf->log, "Sending ping to [%s]",
  286. Address_toString(qp->target, qp->alloc)->bytes);
  287. Dict_putStringCC(dict, "q", "pn", qp->alloc);
  288. BoilerplateResponder_addBoilerplate(pf->br, dict, addr, qp->alloc);
  289. int index = Map_OfPromiseByQuery_put(&q, &qp, &pf->queryMap);
  290. usp->mapHandle = pf->queryMap.handles[index];
  291. }
  292. static void sendToSeeder(Message_t* msg, struct SubnodePathfinder_pvt* pf, enum PFChan_Core ev)
  293. {
  294. Allocator_t* alloc = Allocator_child(pf->alloc);
  295. Message_t* msg2 = Message_clone(msg, alloc);
  296. Err_assert(Message_epush32be(msg2, ev));
  297. RTypes_Error_t* err = Iface_send(&pf->seederIf, msg2);
  298. if (err) {
  299. Log_info(pf->log, "Error sending message to Seeder: %s",
  300. Rffi_printError(err, alloc));
  301. }
  302. Allocator_free(alloc);
  303. }
  304. static Iface_DEFUN peer(Message_t* msg, struct SubnodePathfinder_pvt* pf)
  305. {
  306. sendToSeeder(msg, pf, PFChan_Core_PEER);
  307. struct Address addr = {0};
  308. uint32_t metric = addressForNode(&addr, msg);
  309. String* str = Address_toString(&addr, Message_getAlloc(msg));
  310. int index = AddrSet_indexOf(pf->myPeerAddrs, &addr, AddrSet_Match_BOTH);
  311. if (index > -1) {
  312. struct Address* myPeer = AddrSet_get(pf->myPeerAddrs, index);
  313. // Just in case there is a difference
  314. myPeer->protocolVersion = addr.protocolVersion;
  315. } else {
  316. Log_debug(pf->log, "Peer [%s]", str->bytes);
  317. AddrSet_add(pf->myPeerAddrs, &addr, AddrSet_Match_BOTH);
  318. // immediately ping them
  319. pingNode(pf, &addr);
  320. }
  321. ReachabilityCollector_change(pf->pub.rc, &addr);
  322. if ((metric & ~Metric_IC_PEER_MASK) != Metric_IC_PEER) {
  323. } else if ((metric & Metric_IC_PEER_MASK) == Metric_IC_PEER_MASK) {
  324. } else {
  325. ReachabilityCollector_lagSample(pf->pub.rc, addr.path, (metric & Metric_IC_PEER_MASK));
  326. }
  327. // We send this as Metric_PF_PEER because if it happens that this is a really crappy
  328. // peering link, we would prefer to take the route server's advice and bounce the connection
  329. // over a link which makes more sense to use.
  330. return sendNode(msg, &addr, Metric_PF_PEER, PFChan_Pathfinder_NODE, pf);
  331. }
  332. static Iface_DEFUN peerGone(Message_t* msg, struct SubnodePathfinder_pvt* pf)
  333. {
  334. sendToSeeder(msg, pf, PFChan_Core_PEER_GONE);
  335. struct Address addr = {0};
  336. addressForNode(&addr, msg);
  337. AddrSet_remove(pf->myPeerAddrs, &addr, AddrSet_Match_BOTH);
  338. ReachabilityCollector_unreachable(pf->pub.rc, &addr);
  339. return sendNode(msg, &addr, Metric_DEAD_LINK, PFChan_Pathfinder_NODE, pf);
  340. }
  341. static Iface_DEFUN session(Message_t* msg, struct SubnodePathfinder_pvt* pf)
  342. {
  343. struct Address addr = {0};
  344. addressForNode(&addr, msg);
  345. String* str = Address_toString(&addr, Message_getAlloc(msg));
  346. Log_debug(pf->log, "Session [%s]", str->bytes);
  347. //if (addr.protocolVersion) { NodeCache_discoverNode(pf->nc, &addr); }
  348. return NULL;
  349. }
  350. static Iface_DEFUN sessionEnded(Message_t* msg, struct SubnodePathfinder_pvt* pf)
  351. {
  352. struct Address addr = {0};
  353. addressForNode(&addr, msg);
  354. String* str = Address_toString(&addr, Message_getAlloc(msg));
  355. Log_debug(pf->log, "Session ended [%s]", str->bytes);
  356. //NodeCache_forgetNode(pf->nc, &addr);
  357. return NULL;
  358. }
  359. static Iface_DEFUN discoveredPath(Gcc_UNUSED Message_t* msg, Gcc_UNUSED struct SubnodePathfinder_pvt* pf)
  360. {
  361. //struct Address addr = {0};
  362. //addressForNode(&addr, msg);
  363. //Log_debug(pf->log, "discoveredPath(%s)", Address_toString(&addr, Message_getAlloc(msg))->bytes);
  364. //if (addr.protocolVersion) { NodeCache_discoverNode(pf->nc, &addr); }
  365. return NULL;
  366. }
  367. static Iface_DEFUN handlePing(Message_t* msg, struct SubnodePathfinder_pvt* pf)
  368. {
  369. //Log_debug(pf->log, "Received ping");
  370. Err(Message_epush32be(msg, PFChan_Pathfinder_PONG));
  371. return Iface_next(&pf->pub.eventIf, msg);
  372. }
  373. static Iface_DEFUN handlePong(Gcc_UNUSED Message_t* msg, Gcc_UNUSED struct SubnodePathfinder_pvt* pf)
  374. {
  375. //Log_debug(pf->log, "Received pong");
  376. return NULL;
  377. }
  378. static Iface_DEFUN ctrlMsgFromSwitchPinger(Message_t* msg, struct Iface* iface)
  379. {
  380. struct SubnodePathfinder_pvt* pf =
  381. Identity_containerOf(iface, struct SubnodePathfinder_pvt, switchPingerIf);
  382. Err(Message_epush32be(msg, PFChan_Pathfinder_CTRL_SENDMSG));
  383. return Iface_next(&pf->pub.eventIf, msg);
  384. }
  385. static Iface_DEFUN ctrlMsg(Message_t* msg, struct SubnodePathfinder_pvt* pf)
  386. {
  387. return Iface_next(&pf->switchPingerIf, msg);
  388. }
  389. static Iface_DEFUN unsetupSession(Message_t* msg, struct SubnodePathfinder_pvt* pf)
  390. {
  391. struct PFChan_Node node;
  392. Err(Message_epop(msg, &node, PFChan_Node_SIZE));
  393. Assert_true(!Message_getLength(msg));
  394. struct Address addr = {
  395. .protocolVersion = Endian_bigEndianToHost32(node.version_be),
  396. .path = Endian_bigEndianToHost64(node.path_be),
  397. };
  398. Bits_memcpy(addr.ip6.bytes, node.ip6, 16);
  399. Bits_memcpy(addr.key, node.publicKey, 32);
  400. pingNode(pf, &addr);
  401. return NULL;
  402. }
  403. static Iface_DEFUN incomingMsg(Message_t* msg, struct SubnodePathfinder_pvt* pf)
  404. {
  405. return Iface_next(&pf->msgCoreIf, msg);
  406. }
  407. static Iface_DEFUN linkState(Message_t* msg, struct SubnodePathfinder_pvt* pf)
  408. {
  409. while (Message_getLength(msg)) {
  410. struct PFChan_LinkState_Entry lse;
  411. Err(Message_epop(msg, &lse, PFChan_LinkState_Entry_SIZE));
  412. ReachabilityCollector_updateBandwidthAndDrops(
  413. pf->pub.rc,
  414. lse.peerLabel,
  415. lse.sumOfPackets,
  416. lse.sumOfDrops,
  417. lse.sumOfKb);
  418. }
  419. return NULL;
  420. }
  421. static Iface_DEFUN incomingFromMsgCore(Message_t* msg, struct Iface* iface)
  422. {
  423. struct SubnodePathfinder_pvt* pf =
  424. Identity_containerOf(iface, struct SubnodePathfinder_pvt, msgCoreIf);
  425. Assert_true(Message_getLength(msg) >= (RouteHeader_SIZE + DataHeader_SIZE));
  426. struct RouteHeader* rh = (struct RouteHeader*) Message_bytes(msg);
  427. struct DataHeader* dh = (struct DataHeader*) &rh[1];
  428. Assert_true(DataHeader_getContentType(dh) == ContentType_CJDHT);
  429. Assert_true(!Bits_isZero(rh->publicKey, 32));
  430. Assert_true(rh->version_be);
  431. Assert_true(rh->sh.label_be);
  432. Err(Message_epush32be(msg, PFChan_Pathfinder_SENDMSG));
  433. return Iface_next(&pf->pub.eventIf, msg);
  434. }
  435. static Iface_DEFUN incomingFromEventIf(Message_t* msg, struct Iface* eventIf)
  436. {
  437. struct SubnodePathfinder_pvt* pf =
  438. Identity_containerOf(eventIf, struct SubnodePathfinder_pvt, pub.eventIf);
  439. enum PFChan_Core ev = 0;
  440. Err(Message_epop32be(&ev, msg));
  441. if (SubnodePathfinder_pvt_state_INITIALIZING == pf->state) {
  442. Assert_true(ev == PFChan_Core_CONNECT);
  443. return connected(pf, msg);
  444. }
  445. switch (ev) {
  446. case PFChan_Core_SWITCH_ERR: return switchErr(msg, pf);
  447. case PFChan_Core_SEARCH_REQ: return searchReq(msg, pf);
  448. case PFChan_Core_PEER: return peer(msg, pf);
  449. case PFChan_Core_PEER_GONE: return peerGone(msg, pf);
  450. case PFChan_Core_SESSION: return session(msg, pf);
  451. case PFChan_Core_SESSION_ENDED: return sessionEnded(msg, pf);
  452. case PFChan_Core_DISCOVERED_PATH: return discoveredPath(msg, pf);
  453. case PFChan_Core_MSG: return incomingMsg(msg, pf);
  454. case PFChan_Core_PING: return handlePing(msg, pf);
  455. case PFChan_Core_PONG: return handlePong(msg, pf);
  456. case PFChan_Core_CTRL_MSG: return ctrlMsg(msg, pf);
  457. case PFChan_Core_UNSETUP_SESSION: return unsetupSession(msg, pf);
  458. case PFChan_Core_LINK_STATE: return linkState(msg, pf);
  459. default:;
  460. }
  461. Assert_failure("unexpected event [%d]", ev);
  462. }
  463. static Iface_DEFUN incomingFromSeeder(Message_t* msg, struct Iface* seederIf)
  464. {
  465. struct SubnodePathfinder_pvt* pf =
  466. Identity_containerOf(seederIf, struct SubnodePathfinder_pvt, seederIf);
  467. Log_debug(pf->log, "Got message from seeder");
  468. return Iface_next(&pf->pub.eventIf, msg);
  469. }
  470. static void sendEvent(struct SubnodePathfinder_pvt* pf,
  471. enum PFChan_Pathfinder ev,
  472. void* data,
  473. int size)
  474. {
  475. struct Allocator* alloc = Allocator_child(pf->alloc);
  476. Message_t* msg = Message_new(0, 512+size, alloc);
  477. Err_assert(Message_epush(msg, data, size));
  478. Err_assert(Message_epush32be(msg, ev));
  479. Iface_send(&pf->pub.eventIf, msg);
  480. Allocator_free(alloc);
  481. }
  482. void SubnodePathfinder_start(struct SubnodePathfinder* sp)
  483. {
  484. struct SubnodePathfinder_pvt* pf = Identity_check((struct SubnodePathfinder_pvt*) sp);
  485. struct MsgCore* msgCore = pf->msgCore =
  486. MsgCore_new(pf->base, pf->rand, pf->alloc, pf->log, pf->myScheme);
  487. Iface_plumb(&pf->msgCoreIf, &msgCore->interRouterIf);
  488. PingResponder_new(pf->alloc, pf->log, msgCore, pf->br, pf->myScheme);
  489. struct ReachabilityCollector* rc = pf->pub.rc = ReachabilityCollector_new(
  490. pf->alloc, msgCore, pf->log, pf->base, pf->br, pf->myAddress, pf->myScheme, pf->pub.sp);
  491. rc->userData = pf;
  492. rc->onChange = rcChange;
  493. PeeringSeeder_t* ps = pf->pub.ps = PeeringSeeder_new(
  494. pf->pub.sp,
  495. rc,
  496. pf->alloc,
  497. pf->log,
  498. msgCore,
  499. pf->base,
  500. pf->ca);
  501. Iface_plumb(ps->seederIface, &pf->seederIf);
  502. struct SupernodeHunter* snh = pf->pub.snh = SupernodeHunter_new(
  503. pf->alloc, pf->log, pf->base, pf->pub.sp, pf->myPeerAddrs, msgCore, pf->myAddress, rc);
  504. pf->ra = ReachabilityAnnouncer_new(
  505. pf->alloc, pf->log, pf->base, pf->rand, msgCore, snh, pf->privateKey, pf->myScheme, rc, ps);
  506. struct PFChan_Pathfinder_Connect conn = {
  507. .superiority_be = Endian_hostToBigEndian32(1),
  508. .version_be = Endian_hostToBigEndian32(Version_CURRENT_PROTOCOL)
  509. };
  510. CString_safeStrncpy(conn.userAgent, "Cjdns subnode pathfinder", 64);
  511. sendEvent(pf, PFChan_Pathfinder_CONNECT, &conn, PFChan_Pathfinder_Connect_SIZE);
  512. }
  513. static void sendCurrentSupernode(void* vsp)
  514. {
  515. struct SubnodePathfinder_pvt* pf = Identity_check((struct SubnodePathfinder_pvt*) vsp);
  516. struct Allocator* alloc = Allocator_child(pf->alloc);
  517. Message_t* msgToCore = Message_new(0, 512, alloc);
  518. Iface_CALL(sendNode, msgToCore, &pf->pub.snh->snodeAddr, 0, PFChan_Pathfinder_SNODE, pf);
  519. Allocator_free(alloc);
  520. }
  521. struct SubnodePathfinder* SubnodePathfinder_new(struct Allocator* allocator,
  522. struct Log* log,
  523. EventBase_t* base,
  524. struct Random* rand,
  525. struct Address* myAddress,
  526. uint8_t* privateKey,
  527. struct EncodingScheme* myScheme,
  528. Ca_t* ca)
  529. {
  530. struct Allocator* alloc = Allocator_child(allocator);
  531. struct SubnodePathfinder_pvt* pf =
  532. Allocator_calloc(alloc, sizeof(struct SubnodePathfinder_pvt), 1);
  533. Identity_set(pf);
  534. pf->alloc = alloc;
  535. pf->log = log;
  536. pf->base = base;
  537. pf->rand = rand;
  538. pf->myAddress = myAddress;
  539. pf->myPeerAddrs = AddrSet_new(alloc);
  540. pf->pub.eventIf.send = incomingFromEventIf;
  541. pf->msgCoreIf.send = incomingFromMsgCore;
  542. pf->privateKey = privateKey;
  543. pf->queryMap.allocator = Allocator_child(alloc);
  544. pf->myScheme = myScheme;
  545. pf->br = BoilerplateResponder_new(myScheme, alloc);
  546. pf->pub.sp = SwitchPinger_new(base, rand, log, myAddress, alloc);
  547. pf->switchPingerIf.send = ctrlMsgFromSwitchPinger;
  548. Iface_plumb(&pf->switchPingerIf, &pf->pub.sp->controlHandlerIf);
  549. pf->seederIf.send = incomingFromSeeder;
  550. pf->ca = ca;
  551. Timeout_setInterval(sendCurrentSupernode, pf, 3000, base, alloc);
  552. return &pf->pub;
  553. }