Admin.c 19 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543
  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 "admin/Admin.h"
  16. #include "benc/String.h"
  17. #include "benc/Int.h"
  18. #include "benc/Dict.h"
  19. #include "benc/serialization/standard/BencMessageWriter.h"
  20. #include "benc/serialization/standard/BencMessageReader.h"
  21. #include "memory/Allocator.h"
  22. #include "util/Assert.h"
  23. #include "util/Bits.h"
  24. #include "util/Hex.h"
  25. #include "util/log/Log.h"
  26. #include "util/events/Time.h"
  27. #include "util/events/Timeout.h"
  28. #include "util/Identity.h"
  29. #include "util/platform/Sockaddr.h"
  30. #include "util/Defined.h"
  31. #include <crypto_hash_sha256.h>
  32. #include <stdlib.h>
  33. #include <stdio.h>
  34. static String* TYPE = String_CONST_SO("type");
  35. static String* REQUIRED = String_CONST_SO("required");
  36. static String* STRING = String_CONST_SO("String");
  37. static String* INTEGER = String_CONST_SO("Int");
  38. static String* DICT = String_CONST_SO("Dict");
  39. static String* LIST = String_CONST_SO("List");
  40. static String* TXID = String_CONST_SO("txid");
  41. /** Number of milliseconds before a session times out and outgoing messages are failed. */
  42. #define TIMEOUT_MILLISECONDS 30000
  43. /** map values for tracking time of last message by source address */
  44. struct MapValue
  45. {
  46. /** time when the last incoming message was received. */
  47. uint64_t timeOfLastMessage;
  48. /** used to allocate the memory for the key (Sockaddr) and value (this). */
  49. struct Allocator* allocator;
  50. };
  51. //////// generate time-of-last-message-by-address map
  52. #define Map_USE_HASH
  53. #define Map_USE_COMPARATOR
  54. #define Map_NAME LastMessageTimeByAddr
  55. #define Map_KEY_TYPE struct Sockaddr*
  56. #define Map_VALUE_TYPE struct MapValue*
  57. #include "util/Map.h"
  58. static inline uint32_t Map_LastMessageTimeByAddr_hash(struct Sockaddr** key)
  59. {
  60. uint32_t* k = (uint32_t*) *key;
  61. return k[ ((*key)->addrLen / 4)-1 ];
  62. }
  63. static inline int Map_LastMessageTimeByAddr_compare(struct Sockaddr** keyA, struct Sockaddr** keyB)
  64. {
  65. return Bits_memcmp(*keyA, *keyB, (*keyA)->addrLen);
  66. }
  67. /////// end map
  68. struct Function
  69. {
  70. String* name;
  71. Admin_Function call;
  72. void* context;
  73. bool needsAuth;
  74. Dict* args;
  75. };
  76. struct Admin_pvt
  77. {
  78. struct Admin pub;
  79. struct Iface iface;
  80. struct EventBase* eventBase;
  81. struct Function* functions;
  82. int functionCount;
  83. struct Allocator* allocator;
  84. String* password;
  85. struct Log* logger;
  86. struct Map_LastMessageTimeByAddr map;
  87. /** non-null if we are currently in an admin request. */
  88. struct Message* currentRequest;
  89. /** non-zero if this session able to receive asynchronous messages. */
  90. int asyncEnabled;
  91. /** Length of addresses of clients which communicate with admin. */
  92. uint32_t addrLen;
  93. struct Message* tempSendMsg;
  94. Identity
  95. };
  96. static void sendMessage(struct Message* message, struct Sockaddr* dest, struct Admin_pvt* admin)
  97. {
  98. // stack overflow when used with admin logger.
  99. //Log_keys(admin->logger, "sending message to angel [%s]", message->bytes);
  100. Message_push(message, dest, dest->addrLen, NULL);
  101. Iface_send(&admin->iface, message);
  102. }
  103. static void sendBenc(Dict* message,
  104. struct Sockaddr* dest,
  105. struct Allocator* alloc,
  106. struct Admin_pvt* admin)
  107. {
  108. Message_reset(admin->tempSendMsg);
  109. BencMessageWriter_write(message, admin->tempSendMsg, NULL);
  110. struct Message* msg = Message_new(0, admin->tempSendMsg->length + 32, alloc);
  111. Message_push(msg, admin->tempSendMsg->bytes, admin->tempSendMsg->length, NULL);
  112. sendMessage(msg, dest, admin);
  113. }
  114. /**
  115. * If no incoming data has been sent by this address in TIMEOUT_MILLISECONDS
  116. * then Admin_sendMessage() should fail so that it doesn't endlessly send
  117. * udp packets into outer space after a logging client disconnects.
  118. */
  119. static int checkAddress(struct Admin_pvt* admin, int index, uint64_t now)
  120. {
  121. uint64_t diff = now - admin->map.values[index]->timeOfLastMessage;
  122. // check for backwards time
  123. if (diff > TIMEOUT_MILLISECONDS && diff < ((uint64_t)INT64_MAX)) {
  124. Allocator_free(admin->map.values[index]->allocator);
  125. Map_LastMessageTimeByAddr_remove(index, &admin->map);
  126. return -1;
  127. }
  128. return 0;
  129. }
  130. static void clearExpiredAddresses(void* vAdmin)
  131. {
  132. struct Admin_pvt* admin = Identity_check((struct Admin_pvt*) vAdmin);
  133. uint64_t now = Time_currentTimeMilliseconds(admin->eventBase);
  134. int count = 0;
  135. for (int i = admin->map.count - 1; i >= 0; i--) {
  136. if (checkAddress(admin, i, now)) {
  137. count++;
  138. }
  139. }
  140. Log_debug(admin->logger, "Cleared [%d] expired sessions", count);
  141. }
  142. /**
  143. * public function to send responses
  144. */
  145. int Admin_sendMessage(Dict* message, String* txid, struct Admin* adminPub)
  146. {
  147. struct Admin_pvt* admin = Identity_check((struct Admin_pvt*) adminPub);
  148. if (!admin) {
  149. return 0;
  150. }
  151. Identity_check(admin);
  152. Assert_true(txid && txid->len >= admin->addrLen);
  153. struct Sockaddr_storage addr;
  154. Bits_memcpy(&addr, txid->bytes, admin->addrLen);
  155. // if this is an async call, check if we've got any input from that client.
  156. // if the client is nresponsive then fail the call so logs don't get sent
  157. // out forever after a disconnection.
  158. if (!admin->currentRequest) {
  159. struct Sockaddr* addrPtr = (struct Sockaddr*) &addr.addr;
  160. int index = Map_LastMessageTimeByAddr_indexForKey(&addrPtr, &admin->map);
  161. uint64_t now = Time_currentTimeMilliseconds(admin->eventBase);
  162. if (index < 0 || checkAddress(admin, index, now)) {
  163. return Admin_sendMessage_CHANNEL_CLOSED;
  164. }
  165. }
  166. struct Allocator* alloc = Allocator_child(admin->allocator);
  167. // Bounce back the user-supplied txid.
  168. String userTxid = {
  169. .bytes = txid->bytes + admin->addrLen,
  170. .len = txid->len - admin->addrLen
  171. };
  172. if (txid->len > admin->addrLen) {
  173. Dict_putString(message, TXID, &userTxid, alloc);
  174. }
  175. sendBenc(message, &addr.addr, alloc, admin);
  176. Allocator_free(alloc);
  177. return 0;
  178. }
  179. static inline bool authValid(Dict* message, struct Message* messageBytes, struct Admin_pvt* admin)
  180. {
  181. String* cookieStr = Dict_getString(message, String_CONST("cookie"));
  182. uint32_t cookie = (cookieStr != NULL) ? strtoll(cookieStr->bytes, NULL, 10) : 0;
  183. if (!cookie) {
  184. int64_t* cookieInt = Dict_getInt(message, String_CONST("cookie"));
  185. cookie = (cookieInt) ? *cookieInt : 0;
  186. }
  187. uint64_t nowSecs = Time_currentTimeSeconds(admin->eventBase);
  188. String* submittedHash = Dict_getString(message, String_CONST("hash"));
  189. if (cookie > nowSecs || cookie < nowSecs - 20 || !submittedHash || submittedHash->len != 64) {
  190. return false;
  191. }
  192. uint8_t* hashPtr = CString_strstr(messageBytes->bytes, submittedHash->bytes);
  193. if (!hashPtr || !admin->password) {
  194. return false;
  195. }
  196. uint8_t passAndCookie[64];
  197. snprintf((char*) passAndCookie, 64, "%s%u", admin->password->bytes, cookie);
  198. uint8_t hash[32];
  199. crypto_hash_sha256(hash, passAndCookie, CString_strlen((char*) passAndCookie));
  200. Hex_encode(hashPtr, 64, hash, 32);
  201. crypto_hash_sha256(hash, messageBytes->bytes, messageBytes->length);
  202. Hex_encode(hashPtr, 64, hash, 32);
  203. return Bits_memcmp(hashPtr, submittedHash->bytes, 64) == 0;
  204. }
  205. static bool checkArgs(Dict* args,
  206. struct Function* func,
  207. String* txid,
  208. struct Allocator* requestAlloc,
  209. struct Admin_pvt* admin)
  210. {
  211. struct Dict_Entry* entry = *func->args;
  212. String* error = NULL;
  213. while (entry != NULL) {
  214. String* key = (String*) entry->key;
  215. Assert_ifParanoid(entry->val->type == Object_DICT);
  216. Dict* value = entry->val->as.dictionary;
  217. entry = entry->next;
  218. if (*Dict_getInt(value, String_CONST("required")) == 0) {
  219. continue;
  220. }
  221. String* type = Dict_getString(value, String_CONST("type"));
  222. if ((type == STRING && !Dict_getString(args, key))
  223. || (type == DICT && !Dict_getDict(args, key))
  224. || (type == INTEGER && !Dict_getInt(args, key))
  225. || (type == LIST && !Dict_getList(args, key)))
  226. {
  227. error = String_printf(requestAlloc,
  228. "Entry [%s] is required and must be of type [%s]",
  229. key->bytes,
  230. type->bytes);
  231. break;
  232. }
  233. }
  234. if (error) {
  235. Dict d = Dict_CONST(String_CONST("error"), String_OBJ(error), NULL);
  236. Admin_sendMessage(&d, txid, &admin->pub);
  237. }
  238. return !error;
  239. }
  240. static void asyncEnabled(Dict* args, void* vAdmin, String* txid, struct Allocator* requestAlloc)
  241. {
  242. struct Admin_pvt* admin = Identity_check((struct Admin_pvt*) vAdmin);
  243. int64_t enabled = admin->asyncEnabled;
  244. Dict d = Dict_CONST(String_CONST("asyncEnabled"), Int_OBJ(enabled), NULL);
  245. Admin_sendMessage(&d, txid, &admin->pub);
  246. }
  247. #define ENTRIES_PER_PAGE 8
  248. static void availableFunctions(Dict* args, void* vAdmin, String* txid, struct Allocator* tempAlloc)
  249. {
  250. struct Admin_pvt* admin = Identity_check((struct Admin_pvt*) vAdmin);
  251. int64_t* page = Dict_getInt(args, String_CONST("page"));
  252. uint32_t i = (page) ? *page * ENTRIES_PER_PAGE : 0;
  253. Dict* d = Dict_new(tempAlloc);
  254. Dict* functions = Dict_new(tempAlloc);
  255. int count = 0;
  256. for (; i < (uint32_t)admin->functionCount && count++ < ENTRIES_PER_PAGE; i++) {
  257. Dict_putDict(functions, admin->functions[i].name, admin->functions[i].args, tempAlloc);
  258. }
  259. String* more = String_CONST("more");
  260. if (count >= ENTRIES_PER_PAGE) {
  261. Dict_putInt(d, more, 1, tempAlloc);
  262. }
  263. Dict_putDict(d, String_CONST("availableFunctions"), functions, tempAlloc);
  264. Admin_sendMessage(d, txid, &admin->pub);
  265. }
  266. static void handleRequest(Dict* messageDict,
  267. struct Message* message,
  268. struct Sockaddr* src,
  269. struct Allocator* allocator,
  270. struct Admin_pvt* admin)
  271. {
  272. String* query = Dict_getString(messageDict, String_CONST("q"));
  273. if (!query) {
  274. Log_info(admin->logger, "Got a non-query from admin interface");
  275. return;
  276. }
  277. // txid becomes the user supplied txid combined with the channel num.
  278. String* userTxid = Dict_getString(messageDict, TXID);
  279. uint32_t txidlen = ((userTxid) ? userTxid->len : 0) + src->addrLen;
  280. String* txid = String_newBinary(NULL, txidlen, allocator);
  281. Bits_memcpy(txid->bytes, src, src->addrLen);
  282. if (userTxid) {
  283. Bits_memcpy(txid->bytes + src->addrLen, userTxid->bytes, userTxid->len);
  284. }
  285. // If they're asking for a cookie then lets give them one.
  286. String* cookie = String_CONST("cookie");
  287. if (String_equals(query, cookie)) {
  288. //Log_debug(admin->logger, "Got a request for a cookie");
  289. Dict* d = Dict_new(allocator);
  290. char bytes[32];
  291. snprintf(bytes, 32, "%u", (uint32_t) Time_currentTimeSeconds(admin->eventBase));
  292. String* theCookie = &(String) { .len = CString_strlen(bytes), .bytes = bytes };
  293. Dict_putString(d, cookie, theCookie, allocator);
  294. Admin_sendMessage(d, txid, &admin->pub);
  295. return;
  296. }
  297. // If this is a permitted query, make sure the cookie is right.
  298. String* auth = String_CONST("auth");
  299. bool authed = false;
  300. if (String_equals(query, auth)) {
  301. if (!authValid(messageDict, message, admin)) {
  302. Dict* d = Dict_new(allocator);
  303. Dict_putString(d, String_CONST("error"), String_CONST("Auth failed."), allocator);
  304. Admin_sendMessage(d, txid, &admin->pub);
  305. return;
  306. }
  307. query = Dict_getString(messageDict, String_CONST("aq"));
  308. authed = true;
  309. }
  310. // Then sent a valid authed query, lets track their address so they can receive
  311. // asynchronous messages.
  312. int index = Map_LastMessageTimeByAddr_indexForKey(&src, &admin->map);
  313. uint64_t now = Time_currentTimeMilliseconds(admin->eventBase);
  314. admin->asyncEnabled = 1;
  315. if (index >= 0) {
  316. admin->map.values[index]->timeOfLastMessage = now;
  317. } else if (authed) {
  318. struct Allocator* entryAlloc = Allocator_child(admin->allocator);
  319. struct MapValue* mv = Allocator_calloc(entryAlloc, sizeof(struct MapValue), 1);
  320. mv->timeOfLastMessage = now;
  321. mv->allocator = entryAlloc;
  322. struct Sockaddr* storedAddr = Sockaddr_clone(src, entryAlloc);
  323. Map_LastMessageTimeByAddr_put(&storedAddr, &mv, &admin->map);
  324. } else {
  325. admin->asyncEnabled = 0;
  326. }
  327. Dict* args = Dict_getDict(messageDict, String_CONST("args"));
  328. bool noFunctionsCalled = true;
  329. for (int i = 0; i < admin->functionCount; i++) {
  330. if (String_equals(query, admin->functions[i].name)
  331. && (authed || !admin->functions[i].needsAuth))
  332. {
  333. if (checkArgs(args, &admin->functions[i], txid, message->alloc, admin)) {
  334. admin->functions[i].call(args, admin->functions[i].context, txid, message->alloc);
  335. }
  336. noFunctionsCalled = false;
  337. }
  338. }
  339. if (noFunctionsCalled) {
  340. Dict d = Dict_CONST(
  341. String_CONST("error"),
  342. String_OBJ(String_CONST("No functions matched your request, "
  343. "try Admin_availableFunctions()")),
  344. NULL
  345. );
  346. Admin_sendMessage(&d, txid, &admin->pub);
  347. }
  348. return;
  349. }
  350. static void handleMessage(struct Message* message,
  351. struct Sockaddr* src,
  352. struct Allocator* alloc,
  353. struct Admin_pvt* admin)
  354. {
  355. if (Defined(Log_KEYS)) {
  356. uint8_t lastChar = message->bytes[message->length - 1];
  357. message->bytes[message->length - 1] = '\0';
  358. Log_keys(admin->logger, "Got message from [%s] [%s]",
  359. Sockaddr_print(src, alloc), message->bytes);
  360. message->bytes[message->length - 1] = lastChar;
  361. }
  362. // handle non empty message data
  363. if (message->length > Admin_MAX_REQUEST_SIZE) {
  364. #define TOO_BIG "d5:error16:Request too big.e"
  365. #define TOO_BIG_STRLEN (sizeof(TOO_BIG) - 1)
  366. Bits_memcpyConst(message->bytes, TOO_BIG, TOO_BIG_STRLEN);
  367. message->length = TOO_BIG_STRLEN;
  368. sendMessage(message, src, admin);
  369. return;
  370. }
  371. int origMessageLen = message->length;
  372. Dict* messageDict = NULL;
  373. char* err = BencMessageReader_readNoExcept(message, alloc, &messageDict);
  374. if (err) {
  375. Log_warn(admin->logger,
  376. "Unparsable data from [%s] content: [%s] error: [%s]",
  377. Sockaddr_print(src, alloc), message->bytes, err);
  378. return;
  379. }
  380. if (message->length) {
  381. Log_warn(admin->logger,
  382. "Message from [%s] contained garbage after byte [%d] content: [%s]",
  383. Sockaddr_print(src, alloc), message->length, message->bytes);
  384. return;
  385. }
  386. // put the data back in the front of the message because it is used by the auth checker.
  387. Message_shift(message, origMessageLen, NULL);
  388. handleRequest(messageDict, message, src, alloc, admin);
  389. }
  390. static Iface_DEFUN receiveMessage(struct Message* message, struct Iface* iface)
  391. {
  392. struct Admin_pvt* admin = Identity_containerOf(iface, struct Admin_pvt, iface);
  393. Assert_ifParanoid(message->length >= (int)admin->addrLen);
  394. struct Sockaddr_storage addrStore = { .addr = { .addrLen = 0 } };
  395. Message_pop(message, &addrStore, admin->addrLen, NULL);
  396. struct Allocator* alloc = Allocator_child(admin->allocator);
  397. admin->currentRequest = message;
  398. handleMessage(message, &addrStore.addr, alloc, admin);
  399. admin->currentRequest = NULL;
  400. Allocator_free(alloc);
  401. return NULL;
  402. }
  403. void Admin_registerFunctionWithArgCount(char* name,
  404. Admin_Function callback,
  405. void* callbackContext,
  406. bool needsAuth,
  407. struct Admin_FunctionArg* arguments,
  408. int argCount,
  409. struct Admin* adminPub)
  410. {
  411. struct Admin_pvt* admin = Identity_check((struct Admin_pvt*) adminPub);
  412. String* str = String_new(name, admin->allocator);
  413. admin->functions =
  414. Allocator_realloc(admin->allocator,
  415. admin->functions,
  416. sizeof(struct Function) * (admin->functionCount + 1));
  417. struct Function* fu = &admin->functions[admin->functionCount];
  418. admin->functionCount++;
  419. fu->name = str;
  420. fu->call = callback;
  421. fu->context = callbackContext;
  422. fu->needsAuth = needsAuth;
  423. fu->args = Dict_new(admin->allocator);
  424. for (int i = 0; arguments && i < argCount; i++) {
  425. // "type" must be one of: [ "String", "Int", "Dict", "List" ]
  426. String* type = NULL;
  427. if (!CString_strcmp(arguments[i].type, STRING->bytes)) {
  428. type = STRING;
  429. } else if (!CString_strcmp(arguments[i].type, INTEGER->bytes)) {
  430. type = INTEGER;
  431. } else if (!CString_strcmp(arguments[i].type, DICT->bytes)) {
  432. type = DICT;
  433. } else if (!CString_strcmp(arguments[i].type, LIST->bytes)) {
  434. type = LIST;
  435. } else {
  436. abort();
  437. }
  438. Dict* arg = Dict_new(admin->allocator);
  439. Dict_putString(arg, TYPE, type, admin->allocator);
  440. Dict_putInt(arg, REQUIRED, arguments[i].required, admin->allocator);
  441. String* name = String_new(arguments[i].name, admin->allocator);
  442. Dict_putDict(fu->args, name, arg, admin->allocator);
  443. }
  444. }
  445. struct Admin* Admin_new(struct AddrIface* ai,
  446. struct Log* logger,
  447. struct EventBase* eventBase,
  448. String* password)
  449. {
  450. struct Allocator* alloc = ai->alloc;
  451. struct Admin_pvt* admin = Allocator_calloc(alloc, sizeof(struct Admin_pvt), 1);
  452. Identity_set(admin);
  453. admin->allocator = alloc;
  454. admin->logger = logger;
  455. admin->eventBase = eventBase;
  456. admin->addrLen = ai->addr->addrLen;
  457. admin->map.allocator = alloc;
  458. admin->iface.send = receiveMessage;
  459. Iface_plumb(&admin->iface, &ai->iface);
  460. admin->tempSendMsg = Message_new(0, Admin_MAX_RESPONSE_SIZE, alloc);
  461. admin->password = String_clone(password, alloc);
  462. Timeout_setInterval(clearExpiredAddresses, admin, TIMEOUT_MILLISECONDS * 3, eventBase, alloc);
  463. Admin_registerFunction("Admin_asyncEnabled", asyncEnabled, admin, false, NULL, &admin->pub);
  464. Admin_registerFunction("Admin_availableFunctions", availableFunctions, admin, false,
  465. ((struct Admin_FunctionArg[]) {
  466. { .name = "page", .required = 0, .type = "Int" }
  467. }), &admin->pub);
  468. return &admin->pub;
  469. }