1
0

Admin.c 19 KB

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