MsgCore.c 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359
  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 "benc/Dict.h"
  16. #include "crypto/AddressCalc.h"
  17. #include "memory/Allocator.h"
  18. #include "dht/Address.h"
  19. #include "dht/CJDHTConstants.h"
  20. #include "util/Pinger.h"
  21. #include "subnode/MsgCore.h"
  22. #include "benc/serialization/standard/BencMessageReader.h"
  23. #include "benc/serialization/standard/BencMessageWriter.h"
  24. #include "switch/EncodingScheme.h"
  25. #include "util/Escape.h"
  26. #include "util/Defined.h"
  27. #include "wire/Message.h"
  28. #include "wire/DataHeader.h"
  29. #include "wire/RouteHeader.h"
  30. #include "wire/Error.h"
  31. #define DEFAULT_TIMEOUT_MILLISECONDS 6000
  32. struct ReplyContext
  33. {
  34. struct Address* src;
  35. Dict* content;
  36. Message_t* msg;
  37. Identity
  38. };
  39. struct QueryHandler
  40. {
  41. struct MsgCore_Handler pub;
  42. String* queryType;
  43. struct Allocator* alloc;
  44. struct MsgCore_pvt* mcp;
  45. Identity
  46. };
  47. #define ArrayList_TYPE struct QueryHandler
  48. #define ArrayList_NAME OfQueryHandlers
  49. #include "util/ArrayList.h"
  50. struct MsgCore_pvt
  51. {
  52. struct MsgCore pub;
  53. struct ArrayList_OfQueryHandlers* qh;
  54. struct Pinger* pinger;
  55. struct Log* log;
  56. String* schemeDefinition;
  57. struct EncodingScheme* scheme;
  58. /** Hack hack hack: This should be passed through Pinger.h but the API doesn't exist. */
  59. struct ReplyContext* currentReply;
  60. Identity
  61. };
  62. struct MsgCore_Promise_pvt
  63. {
  64. struct MsgCore_Promise pub;
  65. struct MsgCore_pvt* mcp;
  66. struct Pinger_Ping* ping;
  67. Identity
  68. };
  69. static Iface_DEFUN replyMsg(struct MsgCore_pvt* mcp,
  70. Dict* content,
  71. struct Address* src,
  72. Message_t* msg)
  73. {
  74. Log_debug(mcp->log, "Got reply from [%s]", Address_toString(src, Message_getAlloc(msg))->bytes);
  75. String* txid = Dict_getStringC(content, "txid");
  76. if (!txid) {
  77. Log_debug(mcp->log, "DROP Message with no txid");
  78. return Error(msg, "INVALID");
  79. }
  80. struct ReplyContext* rc = Allocator_calloc(Message_getAlloc(msg), sizeof(struct ReplyContext), 1);
  81. rc->src = src;
  82. rc->content = content;
  83. rc->msg = msg;
  84. Identity_set(rc);
  85. Assert_true(!mcp->currentReply);
  86. mcp->currentReply = rc;
  87. // Pops out in pingerOnResponse() if the reply is indeed valid...
  88. Pinger_pongReceived(txid, mcp->pinger);
  89. mcp->currentReply = NULL;
  90. return NULL;
  91. }
  92. static void pingerOnResponse(String* data, uint32_t milliseconds, void* context)
  93. {
  94. struct MsgCore_Promise_pvt* pp = Identity_check((struct MsgCore_Promise_pvt*) context);
  95. struct MsgCore_pvt* mcp = Identity_check(pp->mcp);
  96. struct ReplyContext* rc = NULL;
  97. if (mcp->currentReply) {
  98. rc = Identity_check(mcp->currentReply);
  99. }
  100. pp->pub.lag = milliseconds;
  101. if (pp->pub.cb) {
  102. pp->pub.cb((rc) ? rc->content : NULL,
  103. (rc) ? rc->src : NULL,
  104. &pp->pub);
  105. }
  106. }
  107. static void sendMsg(struct MsgCore_pvt* mcp,
  108. Dict* msgDict,
  109. struct Address* addr,
  110. struct Allocator* allocator)
  111. {
  112. struct Allocator* alloc = Allocator_child(allocator);
  113. // Send the encoding scheme definition
  114. Dict_putString(msgDict, CJDHTConstants_ENC_SCHEME, mcp->schemeDefinition, allocator);
  115. // And tell the asker which interface the message came from
  116. int encIdx = EncodingScheme_getFormNum(mcp->scheme, addr->path);
  117. Assert_true(encIdx != EncodingScheme_getFormNum_INVALID);
  118. Dict_putInt(msgDict, CJDHTConstants_ENC_INDEX, encIdx, allocator);
  119. // send the protocol version
  120. Dict_putInt(msgDict, CJDHTConstants_PROTOCOL, Version_CURRENT_PROTOCOL, allocator);
  121. String* q = Dict_getStringC(msgDict, "q");
  122. String* sq = Dict_getStringC(msgDict, "sq");
  123. if (q || sq) {
  124. Log_debug(mcp->log, "Send query [%s] to [%s]",
  125. ((q) ? q->bytes : sq->bytes),
  126. Address_toString(addr, alloc)->bytes);
  127. String* txid = Dict_getStringC(msgDict, "txid");
  128. Assert_true(txid);
  129. String* newTxid = String_newBinary(NULL, txid->len + 2, alloc);
  130. Bits_memcpy(&newTxid->bytes[2], txid->bytes, txid->len);
  131. // Always direct queries to the old pathfinder
  132. newTxid->bytes[0] = '0';
  133. newTxid->bytes[1] = '1';
  134. Dict_putStringC(msgDict, "txid", newTxid, alloc);
  135. }
  136. Message_t* msg = Message_new(0, 2048, alloc);
  137. Err_assert(BencMessageWriter_write(msgDict, msg));
  138. //Log_debug(mcp->log, "Sending msg [%s]", Escape_getEscaped(msg->bytes, Message_getLength(msg), alloc));
  139. // Sanity check (make sure the addr was actually calculated)
  140. Assert_true(AddressCalc_validAddress(addr->ip6.bytes));
  141. struct DataHeader data;
  142. Bits_memset(&data, 0, sizeof(struct DataHeader));
  143. DataHeader_setVersion(&data, DataHeader_CURRENT_VERSION);
  144. DataHeader_setContentType(&data, ContentType_CJDHT);
  145. Err_assert(Message_epush(msg, &data, sizeof(struct DataHeader)));
  146. struct RouteHeader route;
  147. Bits_memset(&route, 0, sizeof(struct RouteHeader));
  148. Bits_memcpy(route.ip6, addr->ip6.bytes, 16);
  149. route.version_be = Endian_hostToBigEndian32(addr->protocolVersion);
  150. route.sh.label_be = Endian_hostToBigEndian64(addr->path);
  151. Assert_true(route.sh.label_be != 0xffffffffffffffffull);
  152. route.flags |= RouteHeader_flags_PATHFINDER;
  153. Bits_memcpy(route.publicKey, addr->key, 32);
  154. Err_assert(Message_epush(msg, &route, sizeof(struct RouteHeader)));
  155. Iface_send(&mcp->pub.interRouterIf, msg);
  156. }
  157. void MsgCore_sendResponse(struct MsgCore* core,
  158. Dict* msgDict,
  159. struct Address* target,
  160. struct Allocator* alloc)
  161. {
  162. struct MsgCore_pvt* mcp = Identity_check((struct MsgCore_pvt*) core);
  163. sendMsg(mcp, msgDict, target, alloc);
  164. }
  165. static void pingerSendPing(String* data, void* context)
  166. {
  167. struct MsgCore_Promise_pvt* pp = Identity_check((struct MsgCore_Promise_pvt*) context);
  168. Assert_true(pp->pub.target);
  169. Assert_true(pp->pub.msg);
  170. Dict_putStringC(pp->pub.msg, "txid", data, pp->pub.alloc);
  171. sendMsg(pp->mcp, pp->pub.msg, pp->pub.target, pp->pub.alloc);
  172. }
  173. struct MsgCore_Promise* MsgCore_createQuery(struct MsgCore* core,
  174. uint32_t timeoutMilliseconds,
  175. struct Allocator* allocator)
  176. {
  177. struct MsgCore_pvt* mcp = Identity_check((struct MsgCore_pvt*) core);
  178. if (!timeoutMilliseconds) {
  179. timeoutMilliseconds = DEFAULT_TIMEOUT_MILLISECONDS;
  180. }
  181. struct Pinger_Ping* p = Pinger_newPing(
  182. NULL, pingerOnResponse, pingerSendPing, timeoutMilliseconds, allocator, mcp->pinger);
  183. struct MsgCore_Promise_pvt* out =
  184. Allocator_calloc(p->pingAlloc, sizeof(struct MsgCore_Promise_pvt), 1);
  185. Identity_set(out);
  186. p->context = out;
  187. out->pub.alloc = p->pingAlloc;
  188. out->mcp = mcp;
  189. out->ping = p;
  190. return &out->pub;
  191. }
  192. static struct QueryHandler* getQueryHandler(struct MsgCore_pvt* mcp, String* q)
  193. {
  194. for (int i = 0; i < mcp->qh->length; i++) {
  195. struct QueryHandler* qhx = ArrayList_OfQueryHandlers_get(mcp->qh, i);
  196. Identity_check(qhx);
  197. if (String_equals(qhx->queryType, q)) {
  198. return qhx;
  199. }
  200. }
  201. return NULL;
  202. }
  203. static Iface_DEFUN queryMsg(struct MsgCore_pvt* mcp,
  204. Dict* content,
  205. struct Address* src,
  206. Message_t* msg)
  207. {
  208. String* q = Dict_getStringC(content, "q");
  209. struct QueryHandler* qh = getQueryHandler(mcp, q);
  210. if (!qh) {
  211. qh = getQueryHandler(mcp, String_CONST("pn"));
  212. }
  213. if (!qh) {
  214. Log_debug(mcp->log, "Unhandled query type [%s]", q->bytes);
  215. } else if (!qh->pub.cb) {
  216. Log_info(mcp->log, "Query handler for [%s] not setup", q->bytes);
  217. } else {
  218. return qh->pub.cb(content, src, Message_getAlloc(msg), &qh->pub);
  219. }
  220. return Error(msg, "INVALID");
  221. }
  222. static void qhOnFree(struct Allocator_OnFreeJob* job)
  223. {
  224. struct QueryHandler* qh = Identity_check((struct QueryHandler*) job->userData);
  225. struct MsgCore_pvt* mcp = Identity_check((struct MsgCore_pvt*) qh->mcp);
  226. for (int i = 0; i < mcp->qh->length; i++) {
  227. struct QueryHandler* qhx = ArrayList_OfQueryHandlers_get(mcp->qh, i);
  228. if (qhx == qh) {
  229. ArrayList_OfQueryHandlers_remove(mcp->qh, i);
  230. return;
  231. }
  232. }
  233. }
  234. struct MsgCore_Handler* MsgCore_onQuery(struct MsgCore* core,
  235. char* queryType,
  236. struct Allocator* allocator)
  237. {
  238. struct MsgCore_pvt* mcp = Identity_check((struct MsgCore_pvt*) core);
  239. struct Allocator* alloc = Allocator_child(allocator);
  240. struct QueryHandler* qh = Allocator_calloc(alloc, sizeof(struct QueryHandler), 1);
  241. qh->queryType = String_new(queryType, alloc);
  242. qh->alloc = alloc;
  243. qh->mcp = mcp;
  244. Identity_set(qh);
  245. ArrayList_OfQueryHandlers_add(mcp->qh, qh);
  246. Allocator_onFree(alloc, qhOnFree, qh);
  247. return &qh->pub;
  248. }
  249. static Iface_DEFUN incoming(Message_t* msg, struct Iface* interRouterIf)
  250. {
  251. struct MsgCore_pvt* mcp =
  252. Identity_containerOf(interRouterIf, struct MsgCore_pvt, pub.interRouterIf);
  253. struct Address addr = { .padding = 0 };
  254. struct RouteHeader* hdr = (struct RouteHeader*) Message_bytes(msg);
  255. Err_assert(Message_eshift(msg, -(RouteHeader_SIZE + DataHeader_SIZE)));
  256. Bits_memcpy(addr.ip6.bytes, hdr->ip6, 16);
  257. Bits_memcpy(addr.key, hdr->publicKey, 32);
  258. addr.protocolVersion = Endian_bigEndianToHost32(hdr->version_be);
  259. addr.path = Endian_bigEndianToHost64(hdr->sh.label_be);
  260. Dict* content = NULL;
  261. uint8_t* msgBytes = Message_bytes(msg);
  262. int length = Message_getLength(msg);
  263. //Log_debug(mcp->log, "Receive msg [%s] from [%s]",
  264. // Escape_getEscaped(msg->bytes, Message_getLength(msg), Message_getAlloc(msg)),
  265. // Address_toString(&addr, Message_getAlloc(msg))->bytes);
  266. //
  267. BencMessageReader_readNoExcept(msg, Message_getAlloc(msg), &content);
  268. if (!content) {
  269. char* esc = Escape_getEscaped(msgBytes, length, Message_getAlloc(msg));
  270. Log_debug(mcp->log, "DROP Malformed message [%s]", esc);
  271. return Error(msg, "INVALID");
  272. }
  273. int64_t* verP = Dict_getIntC(content, "p");
  274. if (!verP) {
  275. Log_debug(mcp->log, "DROP Message without version");
  276. return Error(msg, "INVALID");
  277. }
  278. addr.protocolVersion = *verP;
  279. if (!addr.protocolVersion) {
  280. Log_debug(mcp->log, "DROP Message with zero version");
  281. return Error(msg, "INVALID");
  282. }
  283. String* q = Dict_getStringC(content, "q");
  284. String* txid = Dict_getStringC(content, "txid");
  285. if (!txid || !txid->len) {
  286. Log_debug(mcp->log, "Message with no txid [%s]", q ? (q->bytes) : "(no query)");
  287. return Error(msg, "INVALID");
  288. }
  289. if (q) {
  290. if (Defined(SUBNODE)) {
  291. return queryMsg(mcp, content, &addr, msg);
  292. } else {
  293. // Let the old pathfinder handle every query if it is present
  294. return NULL;
  295. }
  296. } else if (txid->len >= 2 && txid->bytes[0] == '0' && txid->bytes[1] == '1') {
  297. txid->bytes = &txid->bytes[2];
  298. txid->len -= 2;
  299. return replyMsg(mcp, content, &addr, msg);
  300. }
  301. return Error(msg, "INVALID");
  302. }
  303. struct MsgCore* MsgCore_new(EventBase_t* base,
  304. struct Random* rand,
  305. struct Allocator* allocator,
  306. struct Log* log,
  307. struct EncodingScheme* scheme)
  308. {
  309. struct Allocator* alloc = Allocator_child(allocator);
  310. struct MsgCore_pvt* mcp = Allocator_calloc(alloc, sizeof(struct MsgCore_pvt), 1);
  311. Identity_set(mcp);
  312. mcp->pub.interRouterIf.send = incoming;
  313. mcp->qh = ArrayList_OfQueryHandlers_new(alloc);
  314. mcp->pinger = Pinger_new(base, rand, log, alloc);
  315. mcp->log = log;
  316. mcp->scheme = scheme;
  317. mcp->schemeDefinition = EncodingScheme_serialize(scheme, alloc);
  318. return &mcp->pub;
  319. }