1
0

EventEmitter.c 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374
  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 "memory/Allocator.h"
  16. #include "wire/PFChan.h"
  17. #include "net/EventEmitter.h"
  18. #include "util/Identity.h"
  19. #include "util/log/Log.h"
  20. #include "wire/Error.h"
  21. #include <stdbool.h>
  22. #define ArrayList_TYPE struct Iface
  23. #define ArrayList_NAME Ifaces
  24. #include "util/ArrayList.h"
  25. #define PING_MAGIC 0x01234567
  26. struct Pathfinder
  27. {
  28. struct EventEmitter_pvt* ee;
  29. struct Iface iface;
  30. struct Allocator* alloc;
  31. uint8_t userAgent[64];
  32. uint32_t superiority;
  33. uint32_t version;
  34. #define Pathfinder_state_DISCONNECTED 0
  35. #define Pathfinder_state_CONNECTED 1
  36. #define Pathfinder_state_ERROR 2
  37. uint16_t state;
  38. uint16_t pathfinderId;
  39. /**
  40. * Number of bytes sent since the last ping, each ping contains this number at the time
  41. * of pinging, then the number is incremented as more messages are sent, if it goes over
  42. * 65535, the state is set to error.
  43. */
  44. uint32_t bytesSinceLastPing;
  45. Identity
  46. };
  47. #define ArrayList_TYPE struct Pathfinder
  48. #define ArrayList_NAME Pathfinders
  49. #include "util/ArrayList.h"
  50. struct EventEmitter_pvt
  51. {
  52. struct EventEmitter pub;
  53. struct Iface trickIf;
  54. struct Allocator* alloc;
  55. struct Log* log;
  56. struct ArrayList_Ifaces* listTable[PFChan_Pathfinder__TOO_HIGH - PFChan_Pathfinder__TOO_LOW];
  57. struct ArrayList_Pathfinders* pathfinders;
  58. uint8_t publicKey[32];
  59. Identity
  60. };
  61. #define OFFSET(ev) (ev - PFChan_Pathfinder__TOO_LOW - 1)
  62. static struct ArrayList_Ifaces* getHandlers(struct EventEmitter_pvt* ee,
  63. enum PFChan_Pathfinder ev,
  64. bool create)
  65. {
  66. if (ev <= PFChan_Pathfinder__TOO_LOW || ev >= PFChan_Pathfinder__TOO_HIGH) { return NULL; }
  67. struct ArrayList_Ifaces* out = ee->listTable[OFFSET(ev)];
  68. if (!out) {
  69. out = ee->listTable[OFFSET(ev)] = ArrayList_Ifaces_new(ee->alloc);
  70. }
  71. return out;
  72. }
  73. static Iface_DEFUN sendToPathfinder(struct Message* msg, struct Pathfinder* pf)
  74. {
  75. if (!pf || pf->state != Pathfinder_state_CONNECTED) { return NULL; }
  76. if (pf->bytesSinceLastPing < 8192 && pf->bytesSinceLastPing + Message_getLength(msg) >= 8192) {
  77. struct Message* ping = Message_new(0, 512, Message_getAlloc(msg));
  78. Er_assert(Message_epush32be(ping, pf->bytesSinceLastPing));
  79. Er_assert(Message_epush32be(ping, PING_MAGIC));
  80. Er_assert(Message_epush32be(ping, PFChan_Core_PING));
  81. Iface_send(&pf->iface, ping);
  82. }
  83. pf->bytesSinceLastPing += Message_getLength(msg);
  84. return Iface_next(&pf->iface, msg);
  85. }
  86. static bool PFChan_Pathfinder_sizeOk(enum PFChan_Pathfinder ev, int size)
  87. {
  88. switch (ev) {
  89. case PFChan_Pathfinder_CONNECT:
  90. return (size == 8 + PFChan_Pathfinder_Connect_SIZE);
  91. case PFChan_Pathfinder_SUPERIORITY:
  92. return (size == 8 + PFChan_Pathfinder_Superiority_SIZE);
  93. case PFChan_Pathfinder_NODE:
  94. case PFChan_Pathfinder_SNODE:
  95. return (size == 8 + PFChan_Node_SIZE);
  96. case PFChan_Pathfinder_SENDMSG:
  97. return (size >= 8 + PFChan_Msg_MIN_SIZE);
  98. case PFChan_Pathfinder_PING:
  99. case PFChan_Pathfinder_PONG:
  100. return (size >= 8 + PFChan_Ping_SIZE);
  101. case PFChan_Pathfinder_SESSIONS:
  102. case PFChan_Pathfinder_PEERS:
  103. case PFChan_Pathfinder_PATHFINDERS:
  104. return (size == 8);
  105. case PFChan_Pathfinder_CTRL_SENDMSG:
  106. return (size >= 8 + PFChan_CtrlMsg_MIN_SIZE);
  107. default:;
  108. }
  109. Assert_failure("invalid event [%d]", ev);
  110. }
  111. // Forget to add the event here? :)
  112. Assert_compileTime(PFChan_Pathfinder__TOO_LOW == 511);
  113. Assert_compileTime(PFChan_Pathfinder__TOO_HIGH == 523);
  114. static bool PFChan_Core_sizeOk(enum PFChan_Core ev, int size)
  115. {
  116. switch (ev) {
  117. case PFChan_Core_CONNECT:
  118. return (size == 8 + PFChan_Core_Connect_SIZE);
  119. case PFChan_Core_PATHFINDER:
  120. case PFChan_Core_PATHFINDER_GONE:
  121. return (size == 8 + PFChan_Core_Pathfinder_SIZE);
  122. case PFChan_Core_SWITCH_ERR:
  123. return (size >= 8 + PFChan_Core_SwitchErr_MIN_SIZE);
  124. case PFChan_Core_SEARCH_REQ:
  125. return (size == 8 + PFChan_Core_SearchReq_SIZE);
  126. case PFChan_Core_PEER:
  127. case PFChan_Core_PEER_GONE:
  128. case PFChan_Core_SESSION:
  129. case PFChan_Core_SESSION_ENDED:
  130. case PFChan_Core_DISCOVERED_PATH:
  131. case PFChan_Core_UNSETUP_SESSION:
  132. return (size == 8 + PFChan_Node_SIZE);
  133. case PFChan_Core_MSG:
  134. return (size >= 8 + PFChan_Msg_MIN_SIZE);
  135. case PFChan_Core_PING:
  136. case PFChan_Core_PONG:
  137. return (size == 8 + PFChan_Ping_SIZE);
  138. case PFChan_Core_CTRL_MSG:
  139. return (size >= 8 + PFChan_CtrlMsg_MIN_SIZE);
  140. case PFChan_Core_LINK_STATE:
  141. return (size >= 8 + PFChan_LinkState_Entry_SIZE) &&
  142. !((size - 8) % PFChan_LinkState_Entry_SIZE);
  143. default:;
  144. }
  145. Assert_failure("invalid event [%d]", ev);
  146. }
  147. // Remember to add the event to this function too!
  148. Assert_compileTime(PFChan_Core__TOO_LOW == 1023);
  149. Assert_compileTime(PFChan_Core__TOO_HIGH == 1040);
  150. static Iface_DEFUN incomingFromCore(struct Message* msg, struct Iface* trickIf)
  151. {
  152. struct EventEmitter_pvt* ee = Identity_containerOf(trickIf, struct EventEmitter_pvt, trickIf);
  153. Assert_true(!((uintptr_t)msg->msgbytes % 4) && "alignment");
  154. enum PFChan_Core ev = Er_assert(Message_epop32be(msg));
  155. Assert_true(PFChan_Core_sizeOk(ev, Message_getLength(msg)+4));
  156. uint32_t pathfinderNum = Er_assert(Message_epop32be(msg));
  157. Er_assert(Message_epush32be(msg, ev));
  158. if (pathfinderNum != 0xffffffff) {
  159. struct Pathfinder* pf = ArrayList_Pathfinders_get(ee->pathfinders, pathfinderNum);
  160. Assert_true(pf && pf->state == Pathfinder_state_CONNECTED);
  161. return sendToPathfinder(msg, pf);
  162. } else {
  163. for (int i = 0; i < ee->pathfinders->length; i++) {
  164. struct Pathfinder* pf = ArrayList_Pathfinders_get(ee->pathfinders, i);
  165. if (!pf || pf->state != Pathfinder_state_CONNECTED) { continue; }
  166. struct Message* messageClone = Message_clone(msg, Message_getAlloc(msg));
  167. Iface_CALL(sendToPathfinder, messageClone, pf);
  168. }
  169. }
  170. return NULL;
  171. }
  172. static struct Message* pathfinderMsg(enum PFChan_Core ev,
  173. struct Pathfinder* pf,
  174. struct Allocator* alloc)
  175. {
  176. struct Message* msg = Message_new(PFChan_Core_Pathfinder_SIZE, 512, alloc);
  177. struct PFChan_Core_Pathfinder* pathfinder = (struct PFChan_Core_Pathfinder*) msg->msgbytes;
  178. pathfinder->superiority_be = Endian_hostToBigEndian32(pf->superiority);
  179. pathfinder->pathfinderId_be = Endian_hostToBigEndian32(pf->pathfinderId);
  180. Bits_memcpy(pathfinder->userAgent, pf->userAgent, 64);
  181. Er_assert(Message_epush32be(msg, 0xffffffff));
  182. Er_assert(Message_epush32be(msg, ev));
  183. return msg;
  184. }
  185. static int handleFromPathfinder(enum PFChan_Pathfinder ev,
  186. struct Message* msg,
  187. struct EventEmitter_pvt* ee,
  188. struct Pathfinder* pf)
  189. {
  190. switch (ev) {
  191. default: return false;
  192. case PFChan_Pathfinder_CONNECT: {
  193. struct PFChan_Pathfinder_Connect connect;
  194. Er_assert(Message_eshift(msg, -8));
  195. Er_assert(Message_epop(msg, &connect, PFChan_Pathfinder_Connect_SIZE));
  196. pf->superiority = Endian_bigEndianToHost32(connect.superiority_be);
  197. pf->version = Endian_bigEndianToHost32(connect.version_be);
  198. Bits_memcpy(pf->userAgent, connect.userAgent, 64);
  199. pf->state = Pathfinder_state_CONNECTED;
  200. struct PFChan_Core_Connect resp;
  201. resp.version_be = Endian_bigEndianToHost32(Version_CURRENT_PROTOCOL);
  202. resp.pathfinderId_be = Endian_hostToBigEndian32(pf->pathfinderId);
  203. Bits_memcpy(resp.publicKey, ee->publicKey, 32);
  204. Er_assert(Message_epush(msg, &resp, PFChan_Core_Connect_SIZE));
  205. Er_assert(Message_epush32be(msg, PFChan_Core_CONNECT));
  206. struct Message* sendMsg = Message_clone(msg, Message_getAlloc(msg));
  207. Iface_CALL(sendToPathfinder, sendMsg, pf);
  208. break;
  209. }
  210. case PFChan_Pathfinder_SUPERIORITY: {
  211. Er_assert(Message_eshift(msg, -8));
  212. pf->superiority = Er_assert(Message_epop32be(msg));
  213. struct Message* resp = pathfinderMsg(PFChan_Core_PATHFINDER, pf, Message_getAlloc(msg));
  214. Iface_CALL(incomingFromCore, resp, &ee->trickIf);
  215. break;
  216. }
  217. case PFChan_Pathfinder_PING: {
  218. struct Message* sendMsg = Message_clone(msg, Message_getAlloc(msg));
  219. Iface_send(&pf->iface, sendMsg);
  220. break;
  221. }
  222. case PFChan_Pathfinder_PONG: {
  223. Er_assert(Message_eshift(msg, -8));
  224. uint32_t cookie = Er_assert(Message_epop32be(msg));
  225. uint32_t count = Er_assert(Message_epop32be(msg));
  226. if (cookie != PING_MAGIC || count > pf->bytesSinceLastPing) {
  227. pf->state = Pathfinder_state_ERROR;
  228. struct Message* resp = pathfinderMsg(PFChan_Core_PATHFINDER_GONE, pf, Message_getAlloc(msg));
  229. Iface_CALL(incomingFromCore, resp, &ee->trickIf);
  230. } else {
  231. pf->bytesSinceLastPing -= count;
  232. }
  233. break;
  234. }
  235. case PFChan_Pathfinder_PATHFINDERS: {
  236. for (int i = 0; i < ee->pathfinders->length; i++) {
  237. struct Pathfinder* xpf = ArrayList_Pathfinders_get(ee->pathfinders, i);
  238. if (!xpf || xpf->state != Pathfinder_state_CONNECTED) { continue; }
  239. struct Allocator* alloc = Allocator_child(Message_getAlloc(msg));
  240. struct Message* resp = pathfinderMsg(PFChan_Core_PATHFINDER, pf, alloc);
  241. Iface_CALL(sendToPathfinder, resp, pf);
  242. Allocator_free(alloc);
  243. }
  244. break;
  245. }
  246. }
  247. return true;
  248. }
  249. static Iface_DEFUN incomingFromPathfinder(struct Message* msg, struct Iface* iface)
  250. {
  251. struct Pathfinder* pf = Identity_containerOf(iface, struct Pathfinder, iface);
  252. struct EventEmitter_pvt* ee = Identity_check((struct EventEmitter_pvt*) pf->ee);
  253. if (Message_getLength(msg) < 4) {
  254. Log_debug(ee->log, "DROPPF runt");
  255. return Error(msg, "RUNT");
  256. }
  257. enum PFChan_Pathfinder ev = Er_assert(Message_epop32be(msg));
  258. Er_assert(Message_epush32be(msg, pf->pathfinderId));
  259. Er_assert(Message_epush32be(msg, ev));
  260. if (ev <= PFChan_Pathfinder__TOO_LOW || ev >= PFChan_Pathfinder__TOO_HIGH) {
  261. Log_debug(ee->log, "DROPPF invalid type [%d]", ev);
  262. return Error(msg, "INVALID");
  263. }
  264. if (!PFChan_Pathfinder_sizeOk(ev, Message_getLength(msg))) {
  265. Log_debug(ee->log, "DROPPF incorrect length[%d] for type [%d]", Message_getLength(msg), ev);
  266. return Error(msg, "INVALID");
  267. }
  268. if (pf->state == Pathfinder_state_DISCONNECTED) {
  269. if (ev != PFChan_Pathfinder_CONNECT) {
  270. Log_debug(ee->log, "DROPPF disconnected and event != CONNECT event:[%d]", ev);
  271. return Error(msg, "INVALID");
  272. }
  273. } else if (pf->state != Pathfinder_state_CONNECTED) {
  274. Log_debug(ee->log, "DROPPF error state");
  275. return Error(msg, "INVALID");
  276. }
  277. if (handleFromPathfinder(ev, msg, ee, pf)) { return NULL; }
  278. struct ArrayList_Ifaces* handlers = getHandlers(ee, ev, false);
  279. if (!handlers) { return NULL; }
  280. for (int i = 0; i < handlers->length; i++) {
  281. struct Message* messageClone = Message_clone(msg, Message_getAlloc(msg));
  282. struct Iface* iface = ArrayList_Ifaces_get(handlers, i);
  283. // We have to call this manually because we don't have an interface handy which is
  284. // actually plumbed with this one.
  285. Assert_true(iface);
  286. Assert_true(iface->send);
  287. Iface_CALL(iface->send, messageClone, iface);
  288. }
  289. return NULL;
  290. }
  291. void EventEmitter_regCore(struct EventEmitter* eventEmitter,
  292. struct Iface* iface,
  293. enum PFChan_Pathfinder ev)
  294. {
  295. struct EventEmitter_pvt* ee = Identity_check((struct EventEmitter_pvt*) eventEmitter);
  296. iface->connectedIf = &ee->trickIf;
  297. Iface_setIdentity(iface);
  298. struct ArrayList_Ifaces* l = getHandlers(ee, ev, true);
  299. if (!l) {
  300. Assert_true(ev == 0);
  301. return;
  302. }
  303. ArrayList_Ifaces_add(l, iface);
  304. }
  305. void EventEmitter_regPathfinderIface(struct EventEmitter* emitter, struct Iface* iface)
  306. {
  307. struct EventEmitter_pvt* ee = Identity_check((struct EventEmitter_pvt*) emitter);
  308. struct Allocator* alloc = Allocator_child(ee->alloc);
  309. struct Pathfinder* pf = Allocator_calloc(alloc, sizeof(struct Pathfinder), 1);
  310. pf->ee = ee;
  311. pf->iface.send = incomingFromPathfinder;
  312. pf->alloc = alloc;
  313. Iface_plumb(&pf->iface, iface);
  314. Identity_set(pf);
  315. int i = 0;
  316. for (; i < ee->pathfinders->length; i++) {
  317. struct Pathfinder* xpf = ArrayList_Pathfinders_get(ee->pathfinders, i);
  318. if (!xpf) { break; }
  319. }
  320. pf->pathfinderId = ArrayList_Pathfinders_put(ee->pathfinders, i, pf);
  321. }
  322. struct EventEmitter* EventEmitter_new(struct Allocator* allocator,
  323. struct Log* log,
  324. uint8_t* publicKey)
  325. {
  326. struct Allocator* alloc = Allocator_child(allocator);
  327. struct EventEmitter_pvt* ee = Allocator_calloc(alloc, sizeof(struct EventEmitter_pvt), 1);
  328. ee->log = log;
  329. ee->alloc = alloc;
  330. ee->trickIf.send = incomingFromCore;
  331. Iface_setIdentity(&ee->trickIf);
  332. ee->pathfinders = ArrayList_Pathfinders_new(ee->alloc);
  333. Bits_memcpy(ee->publicKey, publicKey, 32);
  334. Identity_set(ee);
  335. return &ee->pub;
  336. }