1
0

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. String_newBinary(&txid->bytes[admin->addrLen], txid->len - admin->addrLen, alloc);
  170. if (txid->len > admin->addrLen) {
  171. Dict_putString(message, TXID, userTxid, alloc);
  172. }
  173. sendBenc(message, &addr.addr, alloc, admin);
  174. Dict_remove(message, TXID);
  175. Allocator_free(alloc);
  176. return 0;
  177. }
  178. static inline bool authValid(Dict* message, struct Message* messageBytes, struct Admin_pvt* admin)
  179. {
  180. String* cookieStr = Dict_getString(message, String_CONST("cookie"));
  181. uint32_t cookie = (cookieStr != NULL) ? strtoll(cookieStr->bytes, NULL, 10) : 0;
  182. if (!cookie) {
  183. int64_t* cookieInt = Dict_getInt(message, String_CONST("cookie"));
  184. cookie = (cookieInt) ? *cookieInt : 0;
  185. }
  186. uint64_t nowSecs = Time_currentTimeSeconds(admin->eventBase);
  187. String* submittedHash = Dict_getString(message, String_CONST("hash"));
  188. if (cookie > nowSecs || cookie < nowSecs - 20 || !submittedHash || submittedHash->len != 64) {
  189. return false;
  190. }
  191. uint8_t* hashPtr = CString_strstr(messageBytes->bytes, submittedHash->bytes);
  192. if (!hashPtr || !admin->password) {
  193. return false;
  194. }
  195. uint8_t passAndCookie[64];
  196. snprintf((char*) passAndCookie, 64, "%s%u", admin->password->bytes, cookie);
  197. uint8_t hash[32];
  198. crypto_hash_sha256(hash, passAndCookie, CString_strlen((char*) passAndCookie));
  199. Hex_encode(hashPtr, 64, hash, 32);
  200. crypto_hash_sha256(hash, messageBytes->bytes, messageBytes->length);
  201. Hex_encode(hashPtr, 64, hash, 32);
  202. return Bits_memcmp(hashPtr, submittedHash->bytes, 64) == 0;
  203. }
  204. static bool checkArgs(Dict* args,
  205. struct Function* func,
  206. String* txid,
  207. struct Allocator* requestAlloc,
  208. struct Admin_pvt* admin)
  209. {
  210. struct Dict_Entry* entry = *func->args;
  211. String* error = NULL;
  212. while (entry != NULL) {
  213. String* key = (String*) entry->key;
  214. Assert_ifParanoid(entry->val->type == Object_DICT);
  215. Dict* value = entry->val->as.dictionary;
  216. entry = entry->next;
  217. if (*Dict_getInt(value, String_CONST("required")) == 0) {
  218. continue;
  219. }
  220. String* type = Dict_getString(value, String_CONST("type"));
  221. if ((type == STRING && !Dict_getString(args, key))
  222. || (type == DICT && !Dict_getDict(args, key))
  223. || (type == INTEGER && !Dict_getInt(args, key))
  224. || (type == LIST && !Dict_getList(args, key)))
  225. {
  226. error = String_printf(requestAlloc,
  227. "Entry [%s] is required and must be of type [%s]",
  228. key->bytes,
  229. type->bytes);
  230. break;
  231. }
  232. }
  233. if (error) {
  234. Dict d = Dict_CONST(String_CONST("error"), String_OBJ(error), NULL);
  235. Admin_sendMessage(&d, txid, &admin->pub);
  236. }
  237. return !error;
  238. }
  239. static void asyncEnabled(Dict* args, void* vAdmin, String* txid, struct Allocator* requestAlloc)
  240. {
  241. struct Admin_pvt* admin = Identity_check((struct Admin_pvt*) vAdmin);
  242. int64_t enabled = admin->asyncEnabled;
  243. Dict d = Dict_CONST(String_CONST("asyncEnabled"), Int_OBJ(enabled), NULL);
  244. Admin_sendMessage(&d, txid, &admin->pub);
  245. }
  246. #define ENTRIES_PER_PAGE 8
  247. static void availableFunctions(Dict* args, void* vAdmin, String* txid, struct Allocator* tempAlloc)
  248. {
  249. struct Admin_pvt* admin = Identity_check((struct Admin_pvt*) vAdmin);
  250. int64_t* page = Dict_getInt(args, String_CONST("page"));
  251. uint32_t i = (page) ? *page * ENTRIES_PER_PAGE : 0;
  252. Dict* d = Dict_new(tempAlloc);
  253. Dict* functions = Dict_new(tempAlloc);
  254. int count = 0;
  255. for (; i < (uint32_t)admin->functionCount && count++ < ENTRIES_PER_PAGE; i++) {
  256. Dict_putDict(functions, admin->functions[i].name, admin->functions[i].args, tempAlloc);
  257. }
  258. String* more = String_CONST("more");
  259. if (count >= ENTRIES_PER_PAGE) {
  260. Dict_putInt(d, more, 1, tempAlloc);
  261. }
  262. Dict_putDict(d, String_CONST("availableFunctions"), functions, tempAlloc);
  263. Admin_sendMessage(d, txid, &admin->pub);
  264. }
  265. static void handleRequest(Dict* messageDict,
  266. struct Message* message,
  267. struct Sockaddr* src,
  268. struct Allocator* allocator,
  269. struct Admin_pvt* admin)
  270. {
  271. String* query = Dict_getString(messageDict, String_CONST("q"));
  272. if (!query) {
  273. Log_info(admin->logger, "Got a non-query from admin interface");
  274. return;
  275. }
  276. // txid becomes the user supplied txid combined with the channel num.
  277. String* userTxid = Dict_getString(messageDict, TXID);
  278. uint32_t txidlen = ((userTxid) ? userTxid->len : 0) + src->addrLen;
  279. String* txid = String_newBinary(NULL, txidlen, allocator);
  280. Bits_memcpy(txid->bytes, src, src->addrLen);
  281. if (userTxid) {
  282. Bits_memcpy(txid->bytes + src->addrLen, userTxid->bytes, userTxid->len);
  283. }
  284. // If they're asking for a cookie then lets give them one.
  285. String* cookie = String_CONST("cookie");
  286. if (String_equals(query, cookie)) {
  287. //Log_debug(admin->logger, "Got a request for a cookie");
  288. Dict* d = Dict_new(allocator);
  289. char bytes[32];
  290. snprintf(bytes, 32, "%u", (uint32_t) Time_currentTimeSeconds(admin->eventBase));
  291. String* theCookie = &(String) { .len = CString_strlen(bytes), .bytes = bytes };
  292. Dict_putString(d, cookie, theCookie, allocator);
  293. Admin_sendMessage(d, txid, &admin->pub);
  294. return;
  295. }
  296. // If this is a permitted query, make sure the cookie is right.
  297. String* auth = String_CONST("auth");
  298. bool authed = false;
  299. if (String_equals(query, auth)) {
  300. if (!authValid(messageDict, message, admin)) {
  301. Dict* d = Dict_new(allocator);
  302. Dict_putString(d, String_CONST("error"), String_CONST("Auth failed."), allocator);
  303. Admin_sendMessage(d, txid, &admin->pub);
  304. return;
  305. }
  306. query = Dict_getString(messageDict, String_CONST("aq"));
  307. authed = true;
  308. }
  309. // Then sent a valid authed query, lets track their address so they can receive
  310. // asynchronous messages.
  311. int index = Map_LastMessageTimeByAddr_indexForKey(&src, &admin->map);
  312. uint64_t now = Time_currentTimeMilliseconds(admin->eventBase);
  313. admin->asyncEnabled = 1;
  314. if (index >= 0) {
  315. admin->map.values[index]->timeOfLastMessage = now;
  316. } else if (authed) {
  317. struct Allocator* entryAlloc = Allocator_child(admin->allocator);
  318. struct MapValue* mv = Allocator_calloc(entryAlloc, sizeof(struct MapValue), 1);
  319. mv->timeOfLastMessage = now;
  320. mv->allocator = entryAlloc;
  321. struct Sockaddr* storedAddr = Sockaddr_clone(src, entryAlloc);
  322. Map_LastMessageTimeByAddr_put(&storedAddr, &mv, &admin->map);
  323. } else {
  324. admin->asyncEnabled = 0;
  325. }
  326. Dict* args = Dict_getDict(messageDict, String_CONST("args"));
  327. bool noFunctionsCalled = true;
  328. for (int i = 0; i < admin->functionCount; i++) {
  329. if (String_equals(query, admin->functions[i].name)
  330. && (authed || !admin->functions[i].needsAuth))
  331. {
  332. if (checkArgs(args, &admin->functions[i], txid, message->alloc, admin)) {
  333. admin->functions[i].call(args, admin->functions[i].context, txid, message->alloc);
  334. }
  335. noFunctionsCalled = false;
  336. }
  337. }
  338. if (noFunctionsCalled) {
  339. Dict d = Dict_CONST(
  340. String_CONST("error"),
  341. String_OBJ(String_CONST("No functions matched your request, "
  342. "try Admin_availableFunctions()")),
  343. NULL
  344. );
  345. Admin_sendMessage(&d, txid, &admin->pub);
  346. }
  347. return;
  348. }
  349. static void handleMessage(struct Message* message,
  350. struct Sockaddr* src,
  351. struct Allocator* alloc,
  352. struct Admin_pvt* admin)
  353. {
  354. if (Defined(Log_KEYS)) {
  355. uint8_t lastChar = message->bytes[message->length - 1];
  356. message->bytes[message->length - 1] = '\0';
  357. Log_keys(admin->logger, "Got message from [%s] [%s]",
  358. Sockaddr_print(src, alloc), message->bytes);
  359. message->bytes[message->length - 1] = lastChar;
  360. }
  361. // handle non empty message data
  362. if (message->length > Admin_MAX_REQUEST_SIZE) {
  363. #define TOO_BIG "d5:error16:Request too big.e"
  364. #define TOO_BIG_STRLEN (sizeof(TOO_BIG) - 1)
  365. Bits_memcpyConst(message->bytes, TOO_BIG, TOO_BIG_STRLEN);
  366. message->length = TOO_BIG_STRLEN;
  367. sendMessage(message, src, admin);
  368. return;
  369. }
  370. int origMessageLen = message->length;
  371. Dict* messageDict = NULL;
  372. char* err = BencMessageReader_readNoExcept(message, alloc, &messageDict);
  373. if (err) {
  374. Log_warn(admin->logger,
  375. "Unparsable data from [%s] content: [%s] error: [%s]",
  376. Sockaddr_print(src, alloc), message->bytes, err);
  377. return;
  378. }
  379. if (message->length) {
  380. Log_warn(admin->logger,
  381. "Message from [%s] contained garbage after byte [%d] content: [%s]",
  382. Sockaddr_print(src, alloc), message->length, message->bytes);
  383. return;
  384. }
  385. // put the data back in the front of the message because it is used by the auth checker.
  386. Message_shift(message, origMessageLen, NULL);
  387. handleRequest(messageDict, message, src, alloc, admin);
  388. }
  389. static Iface_DEFUN receiveMessage(struct Message* message, struct Iface* iface)
  390. {
  391. struct Admin_pvt* admin = Identity_containerOf(iface, struct Admin_pvt, iface);
  392. Assert_ifParanoid(message->length >= (int)admin->addrLen);
  393. struct Sockaddr_storage addrStore = { .addr = { .addrLen = 0 } };
  394. Message_pop(message, &addrStore, admin->addrLen, NULL);
  395. struct Allocator* alloc = Allocator_child(admin->allocator);
  396. admin->currentRequest = message;
  397. handleMessage(message, &addrStore.addr, alloc, admin);
  398. admin->currentRequest = NULL;
  399. Allocator_free(alloc);
  400. return NULL;
  401. }
  402. void Admin_registerFunctionWithArgCount(char* name,
  403. Admin_Function callback,
  404. void* callbackContext,
  405. bool needsAuth,
  406. struct Admin_FunctionArg* arguments,
  407. int argCount,
  408. struct Admin* adminPub)
  409. {
  410. struct Admin_pvt* admin = Identity_check((struct Admin_pvt*) adminPub);
  411. String* str = String_new(name, admin->allocator);
  412. admin->functions =
  413. Allocator_realloc(admin->allocator,
  414. admin->functions,
  415. sizeof(struct Function) * (admin->functionCount + 1));
  416. struct Function* fu = &admin->functions[admin->functionCount];
  417. admin->functionCount++;
  418. fu->name = str;
  419. fu->call = callback;
  420. fu->context = callbackContext;
  421. fu->needsAuth = needsAuth;
  422. fu->args = Dict_new(admin->allocator);
  423. for (int i = 0; arguments && i < argCount; i++) {
  424. // "type" must be one of: [ "String", "Int", "Dict", "List" ]
  425. String* type = NULL;
  426. if (!CString_strcmp(arguments[i].type, STRING->bytes)) {
  427. type = STRING;
  428. } else if (!CString_strcmp(arguments[i].type, INTEGER->bytes)) {
  429. type = INTEGER;
  430. } else if (!CString_strcmp(arguments[i].type, DICT->bytes)) {
  431. type = DICT;
  432. } else if (!CString_strcmp(arguments[i].type, LIST->bytes)) {
  433. type = LIST;
  434. } else {
  435. abort();
  436. }
  437. Dict* arg = Dict_new(admin->allocator);
  438. Dict_putString(arg, TYPE, type, admin->allocator);
  439. Dict_putInt(arg, REQUIRED, arguments[i].required, admin->allocator);
  440. String* name = String_new(arguments[i].name, admin->allocator);
  441. Dict_putDict(fu->args, name, arg, admin->allocator);
  442. }
  443. }
  444. struct Admin* Admin_new(struct AddrIface* ai,
  445. struct Log* logger,
  446. struct EventBase* eventBase,
  447. String* password)
  448. {
  449. struct Allocator* alloc = ai->alloc;
  450. struct Admin_pvt* admin = Allocator_calloc(alloc, sizeof(struct Admin_pvt), 1);
  451. Identity_set(admin);
  452. admin->allocator = alloc;
  453. admin->logger = logger;
  454. admin->eventBase = eventBase;
  455. admin->addrLen = ai->addr->addrLen;
  456. admin->map.allocator = alloc;
  457. admin->iface.send = receiveMessage;
  458. Iface_plumb(&admin->iface, &ai->iface);
  459. admin->tempSendMsg = Message_new(0, Admin_MAX_RESPONSE_SIZE, alloc);
  460. admin->password = String_clone(password, alloc);
  461. Timeout_setInterval(clearExpiredAddresses, admin, TIMEOUT_MILLISECONDS * 3, eventBase, alloc);
  462. Admin_registerFunction("Admin_asyncEnabled", asyncEnabled, admin, false, NULL, &admin->pub);
  463. Admin_registerFunction("Admin_availableFunctions", availableFunctions, admin, false,
  464. ((struct Admin_FunctionArg[]) {
  465. { .name = "page", .required = 0, .type = "Int" }
  466. }), &admin->pub);
  467. return &admin->pub;
  468. }