SessionManager.c 23 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594
  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 <http://www.gnu.org/licenses/>.
  14. */
  15. #include "memory/Allocator.h"
  16. #include "wire/PFChan.h"
  17. #include "net/SessionManager.h"
  18. #include "crypto/AddressCalc.h"
  19. #include "util/AddrTools.h"
  20. #include "wire/Error.h"
  21. #include "util/events/Time.h"
  22. #include "util/Defined.h"
  23. #include "wire/RouteHeader.h"
  24. #include "util/events/Timeout.h"
  25. /** Handle numbers 0-3 are reserved for CryptoAuth nonces. */
  26. #define MIN_FIRST_HANDLE 4
  27. #define MAX_FIRST_HANDLE 100000
  28. struct BufferedMessage
  29. {
  30. struct Message* msg;
  31. struct Allocator* alloc;
  32. uint64_t timeSentMilliseconds;
  33. };
  34. struct Ip6 {
  35. uint8_t bytes[16];
  36. };
  37. #define Map_KEY_TYPE struct Ip6
  38. #define Map_VALUE_TYPE struct BufferedMessage*
  39. #define Map_NAME BufferedMessages
  40. #include "util/Map.h"
  41. #define Map_KEY_TYPE struct Ip6
  42. #define Map_VALUE_TYPE struct SessionManager_Session_pvt*
  43. #define Map_NAME OfSessionsByIp6
  44. #define Map_ENABLE_HANDLES
  45. #include "util/Map.h"
  46. struct SessionManager_pvt
  47. {
  48. struct SessionManager pub;
  49. struct Iface eventIf;
  50. struct Allocator* alloc;
  51. struct Map_BufferedMessages bufMap;
  52. struct Map_OfSessionsByIp6 ifaceMap;
  53. struct Log* log;
  54. struct CryptoAuth* cryptoAuth;
  55. struct EventBase* eventBase;
  56. uint32_t firstHandle;
  57. Identity
  58. };
  59. struct SessionManager_Session_pvt
  60. {
  61. struct SessionManager_Session pub;
  62. struct SessionManager_pvt* sessionManager;
  63. struct Allocator* alloc;
  64. Identity
  65. };
  66. #define debugHandlesAndLabel(logger, session, label, message, ...) \
  67. do { \
  68. if (!Defined(Log_DEBUG)) { break; } \
  69. uint8_t path[20]; \
  70. AddrTools_printPath(path, label); \
  71. uint8_t ip[40]; \
  72. AddrTools_printIp(ip, session->pub.caSession->herIp6); \
  73. Log_debug(logger, "ver[%u] send[%d] recv[%u] ip[%s] path[%s] " message, \
  74. session->pub.version, \
  75. session->pub.sendHandle, \
  76. session->pub.receiveHandle, \
  77. ip, \
  78. path, \
  79. __VA_ARGS__); \
  80. } while (0)
  81. //CHECKFILES_IGNORE expecting a ;
  82. #define debugHandlesAndLabel0(logger, session, label, message) \
  83. debugHandlesAndLabel(logger, session, label, "%s", message)
  84. static void sendSession(struct SessionManager_Session_pvt* sess,
  85. uint64_t path,
  86. uint32_t destPf,
  87. enum PFChan_Core ev)
  88. {
  89. struct PFChan_Node session = {
  90. .path_be = Endian_hostToBigEndian64(path),
  91. .metric_be = 0xffffffff,
  92. .version_be = Endian_hostToBigEndian32(sess->pub.version)
  93. };
  94. Bits_memcpyConst(session.ip6, sess->pub.caSession->herIp6, 16);
  95. Bits_memcpyConst(session.publicKey, sess->pub.caSession->herPublicKey, 32);
  96. struct Allocator* alloc = Allocator_child(sess->alloc);
  97. struct Message* msg = Message_new(0, PFChan_Node_SIZE + 512, alloc);
  98. Message_push(msg, &session, PFChan_Node_SIZE, NULL);
  99. Message_push32(msg, destPf, NULL);
  100. Message_push32(msg, ev, NULL);
  101. Iface_send(&sess->sessionManager->eventIf, msg);
  102. Allocator_free(alloc);
  103. }
  104. static inline void check(struct SessionManager_pvt* sm, int mapIndex)
  105. {
  106. Assert_true(sm->ifaceMap.keys[mapIndex].bytes[0] == 0xfc);
  107. uint8_t* herPubKey = sm->ifaceMap.values[mapIndex]->pub.caSession->herPublicKey;
  108. if (!Bits_isZero(herPubKey, 32)) {
  109. uint8_t ip6[16];
  110. AddressCalc_addressForPublicKey(ip6, herPubKey);
  111. Assert_true(!Bits_memcmp(&sm->ifaceMap.keys[mapIndex], ip6, 16));
  112. }
  113. }
  114. static inline struct SessionManager_Session_pvt* sessionForHandle(uint32_t handle,
  115. struct SessionManager_pvt* sm)
  116. {
  117. int index = Map_OfSessionsByIp6_indexForHandle(handle - sm->firstHandle, &sm->ifaceMap);
  118. if (index < 0) { return NULL; }
  119. check(sm, index);
  120. return Identity_check(sm->ifaceMap.values[index]);
  121. }
  122. struct SessionManager_Session* SessionManager_sessionForHandle(uint32_t handle,
  123. struct SessionManager* manager)
  124. {
  125. struct SessionManager_pvt* sm = Identity_check((struct SessionManager_pvt*) manager);
  126. return (struct SessionManager_Session*) sessionForHandle(handle, sm);
  127. }
  128. static inline struct SessionManager_Session_pvt* sessionForIp6(uint8_t ip6[16],
  129. struct SessionManager_pvt* sm)
  130. {
  131. int ifaceIndex = Map_OfSessionsByIp6_indexForKey((struct Ip6*)ip6, &sm->ifaceMap);
  132. if (ifaceIndex == -1) { return NULL; }
  133. check(sm, ifaceIndex);
  134. return Identity_check(sm->ifaceMap.values[ifaceIndex]);
  135. }
  136. struct SessionManager_Session* SessionManager_sessionForIp6(uint8_t* ip6,
  137. struct SessionManager* manager)
  138. {
  139. struct SessionManager_pvt* sm = Identity_check((struct SessionManager_pvt*) manager);
  140. return (struct SessionManager_Session*) sessionForIp6(ip6, sm);
  141. }
  142. struct SessionManager_HandleList* SessionManager_getHandleList(struct SessionManager* manager,
  143. struct Allocator* alloc)
  144. {
  145. struct SessionManager_pvt* sm = Identity_check((struct SessionManager_pvt*) manager);
  146. struct SessionManager_HandleList* out =
  147. Allocator_calloc(alloc, sizeof(struct SessionManager_HandleList), 1);
  148. uint32_t* buff = Allocator_calloc(alloc, 4, sm->ifaceMap.count);
  149. out->length = sm->ifaceMap.count;
  150. out->handles = buff;
  151. for (int i = 0; i < out->length; i++) {
  152. buff[i] = sm->ifaceMap.handles[i] + sm->firstHandle;
  153. }
  154. return out;
  155. }
  156. static struct SessionManager_Session_pvt* getSession(struct SessionManager_pvt* sm,
  157. uint8_t ip6[16],
  158. uint8_t pubKey[32],
  159. uint32_t version,
  160. uint64_t label)
  161. {
  162. struct SessionManager_Session_pvt* sess = sessionForIp6(ip6, sm);
  163. if (sess) {
  164. sess->pub.version = (sess->pub.version) ? sess->pub.version : version;
  165. sess->pub.sendSwitchLabel = (sess->pub.sendSwitchLabel) ? sess->pub.sendSwitchLabel : label;
  166. return sess;
  167. }
  168. struct Allocator* alloc = Allocator_child(sm->alloc);
  169. sess = Allocator_calloc(alloc, sizeof(struct SessionManager_Session_pvt), 1);
  170. Identity_set(sess);
  171. sess->pub.caSession = CryptoAuth_newSession(sm->cryptoAuth, alloc, pubKey, ip6, false, "inner");
  172. int ifaceIndex = Map_OfSessionsByIp6_put((struct Ip6*)ip6, &sess, &sm->ifaceMap);
  173. sess->pub.receiveHandle = sm->ifaceMap.handles[ifaceIndex] + sm->firstHandle;
  174. sess->alloc = alloc;
  175. sess->sessionManager = sm;
  176. sess->pub.version = version;
  177. sess->pub.timeOfLastIn = Time_currentTimeMilliseconds(sm->eventBase);
  178. sess->pub.timeOfLastOut = Time_currentTimeMilliseconds(sm->eventBase);
  179. sess->pub.sendSwitchLabel = label;
  180. //Allocator_onFree(alloc, sessionCleanup, sess);
  181. sendSession(sess, label, 0xffffffff, PFChan_Core_SESSION);
  182. check(sm, ifaceIndex);
  183. return sess;
  184. }
  185. static Iface_DEFUN incomingFromSwitchIf(struct Message* msg, struct Iface* iface)
  186. {
  187. struct SessionManager_pvt* sm =
  188. Identity_containerOf(iface, struct SessionManager_pvt, pub.switchIf);
  189. // SwitchHeader, handle, small cryptoAuth header
  190. if (msg->length < SwitchHeader_SIZE + 4 + 20) {
  191. // This is triggered by Benchmark.c so we really don't want to print log lines constantly.
  192. //Log_debug(sm->log, "DROP runt");
  193. return NULL;
  194. }
  195. struct SwitchHeader* switchHeader = (struct SwitchHeader*) msg->bytes;
  196. Message_shift(msg, -SwitchHeader_SIZE, NULL);
  197. struct SessionManager_Session_pvt* session;
  198. uint32_t nonceOrHandle = Endian_bigEndianToHost32(((uint32_t*)msg->bytes)[0]);
  199. if (nonceOrHandle > 3) {
  200. // > 3 it's a handle.
  201. session = sessionForHandle(nonceOrHandle, sm);
  202. if (!session) {
  203. Log_debug(sm->log, "DROP message with unrecognized handle");
  204. return NULL;
  205. }
  206. Message_shift(msg, -4, NULL);
  207. } else {
  208. // handle + big cryptoauth header
  209. if (msg->length < CryptoHeader_SIZE + 4) {
  210. Log_debug(sm->log, "DROP runt");
  211. return NULL;
  212. }
  213. union CryptoHeader* caHeader = (union CryptoHeader*) msg->bytes;
  214. uint8_t* herKey = caHeader->handshake.publicKey;
  215. uint8_t ip6[16];
  216. // a packet which claims to be "from us" causes problems
  217. if (!AddressCalc_addressForPublicKey(ip6, herKey)) {
  218. Log_debug(sm->log, "DROP Handshake with non-fc key");
  219. return NULL;
  220. }
  221. if (!Bits_memcmp(herKey, sm->cryptoAuth->publicKey, 32)) {
  222. Log_debug(sm->log, "DROP Handshake from 'ourselves'");
  223. return NULL;
  224. }
  225. uint64_t label = Endian_bigEndianToHost64(switchHeader->label_be);
  226. session = getSession(sm, ip6, herKey, 0, label);
  227. debugHandlesAndLabel(sm->log, session, label, "new session nonce[%d]", nonceOrHandle);
  228. }
  229. if (CryptoAuth_decrypt(session->pub.caSession, msg)) {
  230. debugHandlesAndLabel(sm->log, session,
  231. Endian_bigEndianToHost64(switchHeader->label_be),
  232. "DROP Failed decrypting message NoH[%d] state[%s]",
  233. nonceOrHandle,
  234. CryptoAuth_stateString(CryptoAuth_getState(session->pub.caSession)));
  235. return NULL;
  236. }
  237. session->pub.timeOfLastIn = Time_currentTimeMilliseconds(sm->eventBase);
  238. session->pub.bytesIn += msg->length;
  239. bool currentMessageSetup = (nonceOrHandle <= 3);
  240. if (currentMessageSetup) {
  241. session->pub.sendHandle = Message_pop32(msg, NULL);
  242. }
  243. Message_shift(msg, RouteHeader_SIZE, NULL);
  244. struct RouteHeader* header = (struct RouteHeader*) msg->bytes;
  245. if (currentMessageSetup) {
  246. Bits_memcpyConst(&header->sh, switchHeader, SwitchHeader_SIZE);
  247. debugHandlesAndLabel0(sm->log,
  248. session,
  249. Endian_bigEndianToHost64(switchHeader->label_be),
  250. "received start message");
  251. } else {
  252. // RouteHeader is laid out such that no copy of switch header should be needed.
  253. Assert_true(&header->sh == switchHeader);
  254. if (0) { // noisey
  255. debugHandlesAndLabel0(sm->log,
  256. session,
  257. Endian_bigEndianToHost64(switchHeader->label_be),
  258. "received run message");
  259. }
  260. }
  261. header->version_be = Endian_hostToBigEndian32(session->pub.version);
  262. Bits_memcpyConst(header->ip6, session->pub.caSession->herIp6, 16);
  263. Bits_memcpyConst(header->publicKey, session->pub.caSession->herPublicKey, 32);
  264. uint64_t path = Endian_bigEndianToHost64(switchHeader->label_be);
  265. if (!session->pub.sendSwitchLabel) {
  266. session->pub.sendSwitchLabel = path;
  267. }
  268. if (path != session->pub.recvSwitchLabel) {
  269. session->pub.recvSwitchLabel = path;
  270. sendSession(session, path, 0xffffffff, PFChan_Core_DISCOVERED_PATH);
  271. }
  272. return Iface_next(&sm->pub.insideIf, msg);
  273. }
  274. static void checkTimedOutBuffers(struct SessionManager_pvt* sm)
  275. {
  276. for (int i = 0; i < (int)sm->bufMap.count; i++) {
  277. struct BufferedMessage* buffered = sm->bufMap.values[i];
  278. int64_t lag = Time_currentTimeMilliseconds(sm->eventBase) - buffered->timeSentMilliseconds;
  279. if (lag < 10000) { continue; }
  280. Map_BufferedMessages_remove(i, &sm->bufMap);
  281. Allocator_free(buffered->alloc);
  282. i--;
  283. }
  284. }
  285. static void triggerSearch(struct SessionManager_pvt* sm, uint8_t target[16])
  286. {
  287. struct Allocator* eventAlloc = Allocator_child(sm->alloc);
  288. struct Message* eventMsg = Message_new(0, 512, eventAlloc);
  289. Message_push(eventMsg, target, 16, NULL);
  290. Message_push32(eventMsg, 0xffffffff, NULL);
  291. Message_push32(eventMsg, PFChan_Core_SEARCH_REQ, NULL);
  292. Iface_send(&sm->eventIf, eventMsg);
  293. Allocator_free(eventAlloc);
  294. }
  295. static void checkTimedOutSessions(struct SessionManager_pvt* sm)
  296. {
  297. bool searchTriggered = false;
  298. for (int i = 0; i < (int)sm->ifaceMap.count; i++) {
  299. struct SessionManager_Session_pvt* sess = sm->ifaceMap.values[i];
  300. int64_t now = Time_currentTimeMilliseconds(sm->eventBase);
  301. if (now - sess->pub.timeOfLastOut >= sm->pub.sessionIdleAfterMilliseconds &&
  302. now - sess->pub.timeOfLastIn >= sm->pub.sessionIdleAfterMilliseconds)
  303. {
  304. // Session is in idle state
  305. } else if (now - sess->pub.lastSearchTime >= sm->pub.sessionSearchAfterMilliseconds) {
  306. // Session is not in idle state and requires a search
  307. // But we're only going to trigger one search per cycle.
  308. if (searchTriggered) { continue; }
  309. triggerSearch(sm, sess->pub.caSession->herIp6);
  310. sess->pub.lastSearchTime = now;
  311. searchTriggered = true;
  312. }
  313. // Session is in idle state or doesn't need a search right now, check if it's timed out.
  314. if (now - sess->pub.timeOfLastIn < sm->pub.sessionTimeoutMilliseconds) {
  315. sendSession(sess, sess->pub.sendSwitchLabel, 0xffffffff, PFChan_Core_SESSION_ENDED);
  316. Map_OfSessionsByIp6_remove(i, &sm->ifaceMap);
  317. Allocator_free(sess->alloc);
  318. i--;
  319. }
  320. }
  321. }
  322. static void periodically(void* vSessionManager)
  323. {
  324. struct SessionManager_pvt* sm = Identity_check((struct SessionManager_pvt*) vSessionManager);
  325. checkTimedOutSessions(sm);
  326. checkTimedOutBuffers(sm);
  327. }
  328. static void needsLookup(struct SessionManager_pvt* sm, struct Message* msg)
  329. {
  330. struct RouteHeader* header = (struct RouteHeader*) msg->bytes;
  331. if (Defined(Log_DEBUG)) {
  332. uint8_t ipStr[40];
  333. AddrTools_printIp(ipStr, header->ip6);
  334. Log_debug(sm->log, "Buffering a packet to [%s] and beginning a search", ipStr);
  335. }
  336. int index = Map_BufferedMessages_indexForKey((struct Ip6*)header->ip6, &sm->bufMap);
  337. if (index > -1) {
  338. struct BufferedMessage* buffered = sm->bufMap.values[index];
  339. Map_BufferedMessages_remove(index, &sm->bufMap);
  340. Allocator_free(buffered->alloc);
  341. Log_debug(sm->log, "DROP message which needs lookup because new one received");
  342. }
  343. if ((int)sm->bufMap.count >= sm->pub.maxBufferedMessages) {
  344. checkTimedOutBuffers(sm);
  345. if ((int)sm->bufMap.count >= sm->pub.maxBufferedMessages) {
  346. Log_debug(sm->log, "DROP message needing lookup maxBufferedMessages ([%d]) is reached",
  347. sm->pub.maxBufferedMessages);
  348. return;
  349. }
  350. }
  351. struct Allocator* lookupAlloc = Allocator_child(sm->alloc);
  352. struct BufferedMessage* buffered =
  353. Allocator_calloc(lookupAlloc, sizeof(struct BufferedMessage), 1);
  354. buffered->msg = msg;
  355. buffered->alloc = lookupAlloc;
  356. buffered->timeSentMilliseconds = Time_currentTimeMilliseconds(sm->eventBase);
  357. Allocator_adopt(lookupAlloc, msg->alloc);
  358. Assert_true(Map_BufferedMessages_put((struct Ip6*)header->ip6, &buffered, &sm->bufMap) > -1);
  359. triggerSearch(sm, header->ip6);
  360. }
  361. static Iface_DEFUN readyToSend(struct Message* msg,
  362. struct SessionManager_pvt* sm,
  363. struct SessionManager_Session_pvt* sess)
  364. {
  365. struct RouteHeader* header = (struct RouteHeader*) msg->bytes;
  366. Message_shift(msg, -RouteHeader_SIZE, NULL);
  367. struct SwitchHeader* sh;
  368. CryptoAuth_resetIfTimeout(sess->pub.caSession);
  369. if (CryptoAuth_getState(sess->pub.caSession) < CryptoAuth_HANDSHAKE3) {
  370. // Put the handle into the message so that it's authenticated.
  371. Message_push32(msg, sess->pub.receiveHandle, NULL);
  372. // Copy back the SwitchHeader so it is not clobbered.
  373. Message_shift(msg, (CryptoHeader_SIZE + SwitchHeader_SIZE), NULL);
  374. Bits_memcpyConst(msg->bytes, &header->sh, SwitchHeader_SIZE);
  375. sh = (struct SwitchHeader*) msg->bytes;
  376. Message_shift(msg, -(CryptoHeader_SIZE + SwitchHeader_SIZE), NULL);
  377. } else {
  378. sh = &header->sh;
  379. }
  380. // This pointer ceases to be useful.
  381. header = NULL;
  382. sess->pub.timeOfLastOut = Time_currentTimeMilliseconds(sm->eventBase);
  383. sess->pub.bytesOut += msg->length;
  384. Assert_true(!CryptoAuth_encrypt(sess->pub.caSession, msg));
  385. if (CryptoAuth_getState(sess->pub.caSession) >= CryptoAuth_HANDSHAKE3) {
  386. if (0) { // Noisy
  387. debugHandlesAndLabel0(sm->log,
  388. sess,
  389. Endian_bigEndianToHost64(sh->label_be),
  390. "sending run message");
  391. }
  392. Message_push32(msg, sess->pub.sendHandle, NULL);
  393. } else {
  394. debugHandlesAndLabel0(sm->log,
  395. sess,
  396. Endian_bigEndianToHost64(sh->label_be),
  397. "sending start message");
  398. }
  399. // The SwitchHeader should have been moved to the correct location.
  400. Message_shift(msg, SwitchHeader_SIZE, NULL);
  401. Assert_true((uint8_t*)sh == msg->bytes);
  402. return Iface_next(&sm->pub.switchIf, msg);
  403. }
  404. static Iface_DEFUN incomingFromInsideIf(struct Message* msg, struct Iface* iface)
  405. {
  406. struct SessionManager_pvt* sm =
  407. Identity_containerOf(iface, struct SessionManager_pvt, pub.insideIf);
  408. Assert_true(msg->length >= RouteHeader_SIZE);
  409. struct RouteHeader* header = (struct RouteHeader*) msg->bytes;
  410. struct SessionManager_Session_pvt* sess = sessionForIp6(header->ip6, sm);
  411. if (!sess) {
  412. if (!Bits_isZero(header->publicKey, 32) && header->version_be) {
  413. sess = getSession(sm,
  414. header->ip6,
  415. header->publicKey,
  416. Endian_bigEndianToHost32(header->version_be),
  417. Endian_bigEndianToHost64(header->sh.label_be));
  418. } else {
  419. needsLookup(sm, msg);
  420. return NULL;
  421. }
  422. }
  423. if (header->version_be) { sess->pub.version = Endian_bigEndianToHost32(header->version_be); }
  424. if (!sess->pub.version) {
  425. needsLookup(sm, msg);
  426. return NULL;
  427. }
  428. if (header->sh.label_be) {
  429. // fallthrough
  430. } else if (sess->pub.sendSwitchLabel) {
  431. header->sh.label_be = Endian_hostToBigEndian64(sess->pub.sendSwitchLabel);
  432. } else {
  433. needsLookup(sm, msg);
  434. return NULL;
  435. }
  436. return readyToSend(msg, sm, sess);
  437. }
  438. static Iface_DEFUN sessions(struct SessionManager_pvt* sm,
  439. uint32_t sourcePf,
  440. struct Allocator* tempAlloc)
  441. {
  442. for (int i = 0; i < (int)sm->ifaceMap.count; i++) {
  443. struct SessionManager_Session_pvt* sess = sm->ifaceMap.values[i];
  444. sendSession(sess, sess->pub.sendSwitchLabel, sourcePf, PFChan_Core_SESSION);
  445. }
  446. return NULL;
  447. }
  448. static Iface_DEFUN incomingFromEventIf(struct Message* msg, struct Iface* iface)
  449. {
  450. struct SessionManager_pvt* sm = Identity_containerOf(iface, struct SessionManager_pvt, eventIf);
  451. enum PFChan_Pathfinder ev = Message_pop32(msg, NULL);
  452. uint32_t sourcePf = Message_pop32(msg, NULL);
  453. if (ev == PFChan_Pathfinder_SESSIONS) {
  454. Assert_true(!msg->length);
  455. return sessions(sm, sourcePf, msg->alloc);
  456. }
  457. Assert_true(ev == PFChan_Pathfinder_NODE);
  458. struct PFChan_Node node;
  459. Message_pop(msg, &node, PFChan_Node_SIZE, NULL);
  460. Assert_true(!msg->length);
  461. int index = Map_BufferedMessages_indexForKey((struct Ip6*)node.ip6, &sm->bufMap);
  462. struct SessionManager_Session_pvt* sess;
  463. if (index == -1) {
  464. sess = sessionForIp6(node.ip6, sm);
  465. // If we discovered a node we're not interested in ...
  466. if (!sess) { return NULL; }
  467. if (node.metric_be == 0xffffffff) {
  468. // this is a broken path
  469. if (sess->pub.sendSwitchLabel == Endian_bigEndianToHost64(node.path_be)) {
  470. if (sess->pub.sendSwitchLabel == sess->pub.recvSwitchLabel) {
  471. sess->pub.sendSwitchLabel = 0;
  472. } else {
  473. sess->pub.sendSwitchLabel = sess->pub.recvSwitchLabel;
  474. }
  475. }
  476. } else {
  477. sess->pub.sendSwitchLabel = Endian_bigEndianToHost64(node.path_be);
  478. sess->pub.version = Endian_bigEndianToHost32(node.version_be);
  479. }
  480. } else {
  481. sess = getSession(sm,
  482. node.ip6,
  483. node.publicKey,
  484. Endian_bigEndianToHost32(node.version_be),
  485. Endian_bigEndianToHost64(node.path_be));
  486. }
  487. // Send what's on the buffer...
  488. if (index > -1) {
  489. struct BufferedMessage* bm = sm->bufMap.values[index];
  490. Iface_CALL(readyToSend, bm->msg, sm, sess);
  491. Map_BufferedMessages_remove(index, &sm->bufMap);
  492. Allocator_free(bm->alloc);
  493. }
  494. return NULL;
  495. }
  496. struct SessionManager* SessionManager_new(struct Allocator* alloc,
  497. struct EventBase* eventBase,
  498. struct CryptoAuth* cryptoAuth,
  499. struct Random* rand,
  500. struct Log* log,
  501. struct EventEmitter* ee)
  502. {
  503. struct SessionManager_pvt* sm = Allocator_calloc(alloc, sizeof(struct SessionManager_pvt), 1);
  504. sm->alloc = alloc;
  505. sm->pub.switchIf.send = incomingFromSwitchIf;
  506. sm->pub.insideIf.send = incomingFromInsideIf;
  507. sm->bufMap.allocator = alloc;
  508. sm->ifaceMap.allocator = alloc;
  509. sm->log = log;
  510. sm->cryptoAuth = cryptoAuth;
  511. sm->eventBase = eventBase;
  512. sm->pub.sessionTimeoutMilliseconds = SessionManager_SESSION_TIMEOUT_MILLISECONDS_DEFAULT;
  513. sm->pub.maxBufferedMessages = SessionManager_MAX_BUFFERED_MESSAGES_DEFAULT;
  514. sm->pub.sessionIdleAfterMilliseconds = SessionManager_SESSION_IDLE_AFTER_MILLISECONDS_DEFAULT;
  515. sm->pub.sessionSearchAfterMilliseconds =
  516. SessionManager_SESSION_SEARCH_AFTER_MILLISECONDS_DEFAULT;
  517. sm->eventIf.send = incomingFromEventIf;
  518. EventEmitter_regCore(ee, &sm->eventIf, PFChan_Pathfinder_NODE);
  519. EventEmitter_regCore(ee, &sm->eventIf, PFChan_Pathfinder_SESSIONS);
  520. sm->firstHandle =
  521. (Random_uint32(rand) % (MAX_FIRST_HANDLE - MIN_FIRST_HANDLE)) + MIN_FIRST_HANDLE;
  522. Timeout_setInterval(periodically, sm, 10000, eventBase, alloc);
  523. Identity_set(sm);
  524. return &sm->pub;
  525. }