EventEmitter.c 13 KB

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