ktpd_session.c 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916
  1. #include <stdlib.h>
  2. #include <stdio.h>
  3. #include <string.h>
  4. #include <assert.h>
  5. #include <unistd.h>
  6. #include <errno.h>
  7. #include <sys/types.h>
  8. #include <sys/stat.h>
  9. #include <fcntl.h>
  10. #include <sys/socket.h>
  11. #include <sys/un.h>
  12. #include <syslog.h>
  13. #include <poll.h>
  14. #include <sys/wait.h>
  15. #include <faux/str.h>
  16. #include <faux/async.h>
  17. #include <faux/msg.h>
  18. #include <faux/eloop.h>
  19. #include <klish/ksession.h>
  20. #include <klish/ksession_parse.h>
  21. #include <klish/ktp.h>
  22. #include <klish/ktp_session.h>
  23. typedef enum {
  24. KTPD_SESSION_STATE_DISCONNECTED = 'd',
  25. KTPD_SESSION_STATE_UNAUTHORIZED = 'a',
  26. KTPD_SESSION_STATE_IDLE = 'i',
  27. KTPD_SESSION_STATE_WAIT_FOR_PROCESS = 'p',
  28. } ktpd_session_state_e;
  29. struct ktpd_session_s {
  30. ksession_t *session;
  31. ktpd_session_state_e state;
  32. uid_t uid;
  33. gid_t gid;
  34. char *user;
  35. faux_async_t *async;
  36. faux_hdr_t *hdr; // Engine will receive header and then msg
  37. faux_eloop_t *eloop; // External link, dont's free()
  38. kexec_t *exec;
  39. bool_t exit;
  40. };
  41. // Static declarations
  42. static bool_t ktpd_session_read_cb(faux_async_t *async,
  43. faux_buf_t *buf, size_t len, void *user_data);
  44. static bool_t wait_for_actions_ev(faux_eloop_t *eloop, faux_eloop_type_e type,
  45. void *associated_data, void *user_data);
  46. bool_t client_ev(faux_eloop_t *eloop, faux_eloop_type_e type,
  47. void *associated_data, void *user_data);
  48. static bool_t ktpd_session_exec(ktpd_session_t *ktpd, const char *line,
  49. int *retcode, faux_error_t *error, bool_t dry_run);
  50. static bool_t action_stdout_ev(faux_eloop_t *eloop, faux_eloop_type_e type,
  51. void *associated_data, void *user_data);
  52. static bool_t action_stderr_ev(faux_eloop_t *eloop, faux_eloop_type_e type,
  53. void *associated_data, void *user_data);
  54. ktpd_session_t *ktpd_session_new(int sock, kscheme_t *scheme,
  55. const char *start_entry, faux_eloop_t *eloop)
  56. {
  57. ktpd_session_t *ktpd = NULL;
  58. if (sock < 0)
  59. return NULL;
  60. if (!eloop)
  61. return NULL;
  62. ktpd = faux_zmalloc(sizeof(*ktpd));
  63. assert(ktpd);
  64. if (!ktpd)
  65. return NULL;
  66. // Init
  67. ktpd->state = KTPD_SESSION_STATE_IDLE;
  68. ktpd->eloop = eloop;
  69. ktpd->session = ksession_new(scheme, start_entry);
  70. assert(ktpd->session);
  71. ktpd->exec = NULL;
  72. // Exit flag. It differs from ksession done flag because KTPD session
  73. // can't exit immediately. It must finish current command processing
  74. // before really stop the event loop. Note: User defined plugin
  75. // function must use ksession done flag. This exit flag is internal
  76. // feature of KTPD session.
  77. ktpd->exit = BOOL_FALSE;
  78. // Async object
  79. ktpd->async = faux_async_new(sock);
  80. assert(ktpd->async);
  81. // Receive message header first
  82. faux_async_set_read_limits(ktpd->async,
  83. sizeof(faux_hdr_t), sizeof(faux_hdr_t));
  84. faux_async_set_read_cb(ktpd->async, ktpd_session_read_cb, ktpd);
  85. ktpd->hdr = NULL;
  86. faux_async_set_stall_cb(ktpd->async, ktp_stall_cb, ktpd->eloop);
  87. // Eloop callbacks
  88. faux_eloop_add_fd(ktpd->eloop, ktpd_session_fd(ktpd), POLLIN,
  89. client_ev, ktpd);
  90. faux_eloop_add_signal(ktpd->eloop, SIGCHLD, wait_for_actions_ev, ktpd);
  91. return ktpd;
  92. }
  93. void ktpd_session_free(ktpd_session_t *ktpd)
  94. {
  95. if (!ktpd)
  96. return;
  97. kexec_free(ktpd->exec);
  98. ksession_free(ktpd->session);
  99. faux_free(ktpd->hdr);
  100. close(ktpd_session_fd(ktpd));
  101. faux_async_free(ktpd->async);
  102. faux_free(ktpd);
  103. }
  104. static bool_t ktpd_session_process_cmd(ktpd_session_t *ktpd, faux_msg_t *msg)
  105. {
  106. char *line = NULL;
  107. int retcode = -1;
  108. ktp_cmd_e cmd = KTP_CMD_ACK;
  109. faux_error_t *error = NULL;
  110. bool_t rc = BOOL_FALSE;
  111. bool_t dry_run = BOOL_FALSE;
  112. uint32_t status = KTP_STATUS_NONE;
  113. bool_t ret = BOOL_TRUE;
  114. assert(ktpd);
  115. assert(msg);
  116. // Get line from message
  117. if (!(line = faux_msg_get_str_param_by_type(msg, KTP_PARAM_LINE))) {
  118. ktp_send_error(ktpd->async, cmd, "The line is not specified");
  119. return BOOL_FALSE;
  120. }
  121. // Get dry-run flag from message
  122. if (KTP_STATUS_IS_DRY_RUN(faux_msg_get_status(msg)))
  123. dry_run = BOOL_TRUE;
  124. error = faux_error_new();
  125. ktpd->exec = NULL;
  126. rc = ktpd_session_exec(ktpd, line, &retcode, error, dry_run);
  127. faux_str_free(line);
  128. // Command is scheduled. Eloop will wait for ACTION completion.
  129. // So inform client about it and about command features like
  130. // interactive/non-interactive.
  131. if (ktpd->exec) {
  132. faux_msg_t *ack = NULL;
  133. ktp_status_e status = KTP_STATUS_INCOMPLETED;
  134. ack = ktp_msg_preform(cmd, status);
  135. faux_msg_send_async(ack, ktpd->async);
  136. faux_msg_free(ack);
  137. faux_error_free(error);
  138. return BOOL_TRUE; // Continue and wait for ACTION
  139. }
  140. // Here we don't need to wait for the action. We have retcode already.
  141. if (ksession_done(ktpd->session)) {
  142. ktpd->exit = BOOL_TRUE;
  143. status |= KTP_STATUS_EXIT;
  144. }
  145. if (rc) {
  146. uint8_t retcode8bit = 0;
  147. faux_msg_t *ack = ktp_msg_preform(cmd, status);
  148. retcode8bit = (uint8_t)(retcode & 0xff);
  149. faux_msg_add_param(ack, KTP_PARAM_RETCODE, &retcode8bit, 1);
  150. faux_msg_send_async(ack, ktpd->async);
  151. faux_msg_free(ack);
  152. } else {
  153. char *err = faux_error_cstr(error);
  154. ktp_send_error(ktpd->async, cmd, err);
  155. faux_str_free(err);
  156. ret = BOOL_FALSE;
  157. }
  158. faux_error_free(error);
  159. return ret;
  160. }
  161. static bool_t ktpd_session_exec(ktpd_session_t *ktpd, const char *line,
  162. int *retcode, faux_error_t *error, bool_t dry_run)
  163. {
  164. kexec_t *exec = NULL;
  165. assert(ktpd);
  166. if (!ktpd)
  167. return BOOL_FALSE;
  168. // Parsing
  169. exec = ksession_parse_for_exec(ktpd->session, line, error);
  170. if (!exec)
  171. return BOOL_FALSE;
  172. // Set dry-run flag
  173. kexec_set_dry_run(exec, dry_run);
  174. // Session status can be changed while parsing
  175. // NOTE: kexec_t is atomic now
  176. // if (ksession_done(ktpd->session)) {
  177. // kexec_free(exec);
  178. // return BOOL_FALSE; // Because action is not completed
  179. // }
  180. // Execute kexec and then wait for completion using global Eloop
  181. if (!kexec_exec(exec)) {
  182. kexec_free(exec);
  183. return BOOL_FALSE; // Something went wrong
  184. }
  185. // If kexec contains only non-exec (for example dry-run) ACTIONs then
  186. // we don't need event loop and can return here.
  187. if (kexec_retcode(exec, retcode)) {
  188. kexec_free(exec);
  189. return BOOL_TRUE;
  190. }
  191. // Save kexec pointer to use later
  192. ktpd->state = KTPD_SESSION_STATE_WAIT_FOR_PROCESS;
  193. ktpd->exec = exec;
  194. faux_eloop_add_fd(ktpd->eloop, kexec_stdout(exec), POLLIN,
  195. action_stdout_ev, ktpd);
  196. faux_eloop_add_fd(ktpd->eloop, kexec_stderr(exec), POLLIN,
  197. action_stderr_ev, ktpd);
  198. return BOOL_TRUE;
  199. }
  200. static bool_t wait_for_actions_ev(faux_eloop_t *eloop, faux_eloop_type_e type,
  201. void *associated_data, void *user_data)
  202. {
  203. int wstatus = 0;
  204. pid_t child_pid = -1;
  205. ktpd_session_t *ktpd = (ktpd_session_t *)user_data;
  206. int retcode = -1;
  207. uint8_t retcode8bit = 0;
  208. faux_msg_t *ack = NULL;
  209. ktp_cmd_e cmd = KTP_CMD_ACK;
  210. uint32_t status = KTP_STATUS_NONE;
  211. if (!ktpd)
  212. return BOOL_FALSE;
  213. // Wait for any child process. Doesn't block.
  214. while ((child_pid = waitpid(-1, &wstatus, WNOHANG)) > 0) {
  215. if (ktpd->exec)
  216. kexec_continue_command_execution(ktpd->exec, child_pid,
  217. wstatus);
  218. }
  219. if (!ktpd->exec)
  220. return BOOL_TRUE;
  221. // Check if kexec is done now
  222. if (!kexec_retcode(ktpd->exec, &retcode))
  223. return BOOL_TRUE; // Continue
  224. faux_eloop_del_fd(eloop, kexec_stdout(ktpd->exec));
  225. faux_eloop_del_fd(eloop, kexec_stderr(ktpd->exec));
  226. kexec_free(ktpd->exec);
  227. ktpd->exec = NULL;
  228. ktpd->state = KTPD_SESSION_STATE_IDLE;
  229. // All kexec_t actions are done so can break the loop if needed.
  230. if (ksession_done(ktpd->session)) {
  231. ktpd->exit = BOOL_TRUE;
  232. status |= KTP_STATUS_EXIT; // Notify client about exiting
  233. }
  234. // Send ACK message
  235. ack = ktp_msg_preform(cmd, status);
  236. retcode8bit = (uint8_t)(retcode & 0xff);
  237. faux_msg_add_param(ack, KTP_PARAM_RETCODE, &retcode8bit, 1);
  238. faux_msg_send_async(ack, ktpd->async);
  239. faux_msg_free(ack);
  240. type = type; // Happy compiler
  241. associated_data = associated_data; // Happy compiler
  242. if (ktpd->exit)
  243. return BOOL_FALSE;
  244. return BOOL_TRUE;
  245. }
  246. static int compl_compare(const void *first, const void *second)
  247. {
  248. const char *f = (const char *)first;
  249. const char *s = (const char *)second;
  250. return strcmp(f, s);
  251. }
  252. static int compl_kcompare(const void *key, const void *list_item)
  253. {
  254. const char *f = (const char *)key;
  255. const char *s = (const char *)list_item;
  256. return strcmp(f, s);
  257. }
  258. static bool_t ktpd_session_process_completion(ktpd_session_t *ktpd, faux_msg_t *msg)
  259. {
  260. char *line = NULL;
  261. faux_msg_t *ack = NULL;
  262. kpargv_t *pargv = NULL;
  263. ktp_cmd_e cmd = KTP_COMPLETION_ACK;
  264. uint32_t status = KTP_STATUS_NONE;
  265. const char *prefix = NULL;
  266. size_t prefix_len = 0;
  267. assert(ktpd);
  268. assert(msg);
  269. // Get line from message
  270. if (!(line = faux_msg_get_str_param_by_type(msg, KTP_PARAM_LINE))) {
  271. ktp_send_error(ktpd->async, cmd, NULL);
  272. return BOOL_FALSE;
  273. }
  274. // Parsing
  275. pargv = ksession_parse_for_completion(ktpd->session, line);
  276. faux_str_free(line);
  277. if (!pargv) {
  278. ktp_send_error(ktpd->async, cmd, NULL);
  279. return BOOL_FALSE;
  280. }
  281. kpargv_debug(pargv);
  282. if (ksession_done(ktpd->session)) {
  283. ktpd->exit = BOOL_TRUE;
  284. status |= KTP_STATUS_EXIT; // Notify client about exiting
  285. }
  286. // Prepare ACK message
  287. ack = ktp_msg_preform(cmd, status);
  288. // Last unfinished word. Common prefix for all completions
  289. prefix = kpargv_last_arg(pargv);
  290. if (!faux_str_is_empty(prefix)) {
  291. prefix_len = strlen(prefix);
  292. faux_msg_add_param(ack, KTP_PARAM_PREFIX, prefix, prefix_len);
  293. }
  294. // Fill msg with possible completions
  295. if (!kpargv_completions_is_empty(pargv)) {
  296. const kentry_t *candidate = NULL;
  297. kpargv_completions_node_t *citer = kpargv_completions_iter(pargv);
  298. faux_list_node_t *compl_iter = NULL;
  299. faux_list_t *completions = NULL;
  300. char *compl_str = NULL;
  301. completions = faux_list_new(FAUX_LIST_SORTED, FAUX_LIST_UNIQUE,
  302. compl_compare, compl_kcompare,
  303. (void (*)(void *))faux_str_free);
  304. while ((candidate = kpargv_completions_each(&citer))) {
  305. const kentry_t *completion = NULL;
  306. kparg_t *parg = NULL;
  307. int rc = -1;
  308. char *out = NULL;
  309. bool_t res = BOOL_FALSE;
  310. char *l = NULL; // One line of completion
  311. const char *str = NULL;
  312. // Get completion entry from candidate entry
  313. completion = kentry_nested_by_purpose(candidate,
  314. KENTRY_PURPOSE_COMPLETION);
  315. // If candidate entry doesn't contain completion then try
  316. // to get completion from entry's PTYPE
  317. if (!completion) {
  318. const kentry_t *ptype = NULL;
  319. ptype = kentry_nested_by_purpose(candidate,
  320. KENTRY_PURPOSE_PTYPE);
  321. if (!ptype)
  322. continue;
  323. completion = kentry_nested_by_purpose(ptype,
  324. KENTRY_PURPOSE_COMPLETION);
  325. }
  326. if (!completion)
  327. continue;
  328. parg = kparg_new(candidate, prefix);
  329. kpargv_set_candidate_parg(pargv, parg);
  330. res = ksession_exec_locally(ktpd->session, completion,
  331. pargv, &rc, &out);
  332. kparg_free(parg);
  333. if (!res || (rc < 0) || !out)
  334. continue;
  335. // Get all completions one by one
  336. str = out;
  337. while ((l = faux_str_getline(str, &str))) {
  338. // Compare prefix
  339. if ((prefix_len > 0) &&
  340. (faux_str_cmpn(prefix, l, prefix_len) != 0)) {
  341. faux_str_free(l);
  342. continue;
  343. }
  344. compl_str = l + prefix_len;
  345. faux_list_add(completions, faux_str_dup(compl_str));
  346. faux_str_free(l);
  347. }
  348. faux_str_free(out);
  349. }
  350. // Put completion list to message
  351. compl_iter = faux_list_head(completions);
  352. while ((compl_str = faux_list_each(&compl_iter))) {
  353. faux_msg_add_param(ack, KTP_PARAM_LINE,
  354. compl_str, strlen(compl_str));
  355. }
  356. faux_list_free(completions);
  357. }
  358. faux_msg_send_async(ack, ktpd->async);
  359. faux_msg_free(ack);
  360. kpargv_free(pargv);
  361. return BOOL_TRUE;
  362. }
  363. typedef struct help_s {
  364. char *prefix;
  365. char *line;
  366. } help_t;
  367. static int help_compare(const void *first, const void *second)
  368. {
  369. const help_t *f = (const help_t *)first;
  370. const help_t *s = (const help_t *)second;
  371. return strcmp(f->prefix, s->prefix);
  372. }
  373. static int help_kcompare(const void *key, const void *list_item)
  374. {
  375. const char *f = (const char *)key;
  376. const help_t *s = (const help_t *)list_item;
  377. return strcmp(f, s->prefix);
  378. }
  379. static help_t *help_new(char *prefix, char *line)
  380. {
  381. help_t *help = NULL;
  382. help = faux_zmalloc(sizeof(*help));
  383. help->prefix = prefix;
  384. help->line = line;
  385. return help;
  386. }
  387. static void help_free(void *ptr)
  388. {
  389. help_t *help = (help_t *)ptr;
  390. faux_free(help->prefix);
  391. faux_free(help->line);
  392. }
  393. static bool_t ktpd_session_process_help(ktpd_session_t *ktpd, faux_msg_t *msg)
  394. {
  395. char *line = NULL;
  396. faux_msg_t *ack = NULL;
  397. kpargv_t *pargv = NULL;
  398. ktp_cmd_e cmd = KTP_HELP_ACK;
  399. uint32_t status = KTP_STATUS_NONE;
  400. const char *prefix = NULL;
  401. size_t prefix_len = 0;
  402. assert(ktpd);
  403. assert(msg);
  404. // Get line from message
  405. if (!(line = faux_msg_get_str_param_by_type(msg, KTP_PARAM_LINE))) {
  406. ktp_send_error(ktpd->async, cmd, NULL);
  407. return BOOL_FALSE;
  408. }
  409. // Parsing
  410. pargv = ksession_parse_for_completion(ktpd->session, line);
  411. faux_str_free(line);
  412. if (!pargv) {
  413. ktp_send_error(ktpd->async, cmd, NULL);
  414. return BOOL_FALSE;
  415. }
  416. if (ksession_done(ktpd->session)) {
  417. ktpd->exit = BOOL_TRUE;
  418. status |= KTP_STATUS_EXIT; // Notify client about exiting
  419. }
  420. // Prepare ACK message
  421. ack = ktp_msg_preform(cmd, status);
  422. // Last unfinished word. Common prefix for all entries
  423. prefix = kpargv_last_arg(pargv);
  424. if (!faux_str_is_empty(prefix)) {
  425. prefix_len = strlen(prefix);
  426. // faux_msg_add_param(ack, KTP_PARAM_PREFIX, prefix, prefix_len);
  427. }
  428. /*
  429. // Fill msg with possible completions
  430. if (!kpargv_completions_is_empty(pargv)) {
  431. const kentry_t *candidate = NULL;
  432. kpargv_completions_node_t *citer = kpargv_completions_iter(pargv);
  433. faux_list_node_t *help_iter = NULL;
  434. faux_list_t *help_list = NULL;
  435. char *compl_str = NULL;
  436. help_list = faux_list_new(FAUX_LIST_SORTED, FAUX_LIST_UNIQUE,
  437. help_compare, help_kcompare, help_free);
  438. while ((candidate = kpargv_completions_each(&citer))) {
  439. const kentry_t *completion = NULL;
  440. kparg_t *parg = NULL;
  441. int rc = -1;
  442. char *out = NULL;
  443. bool_t res = BOOL_FALSE;
  444. char *l = NULL; // One line of completion
  445. const char *str = NULL;
  446. // Get completion entry from candidate entry
  447. completion = kentry_nested_by_purpose(candidate,
  448. KENTRY_PURPOSE_COMPLETION);
  449. // If candidate entry doesn't contain completion then try
  450. // to get completion from entry's PTYPE
  451. if (!completion) {
  452. const kentry_t *ptype = NULL;
  453. ptype = kentry_nested_by_purpose(candidate,
  454. KENTRY_PURPOSE_PTYPE);
  455. if (!ptype)
  456. continue;
  457. completion = kentry_nested_by_purpose(ptype,
  458. KENTRY_PURPOSE_COMPLETION);
  459. }
  460. if (!completion)
  461. continue;
  462. parg = kparg_new(candidate, prefix);
  463. kpargv_set_candidate_parg(pargv, parg);
  464. res = ksession_exec_locally(ktpd->session, completion,
  465. pargv, &rc, &out);
  466. kparg_free(parg);
  467. if (!res || (rc < 0) || !out)
  468. continue;
  469. // Get all completions one by one
  470. str = out;
  471. while ((l = faux_str_getline(str, &str))) {
  472. // Compare prefix
  473. if ((prefix_len > 0) &&
  474. (faux_str_cmpn(prefix, l, prefix_len) != 0)) {
  475. faux_str_free(l);
  476. continue;
  477. }
  478. compl_str = l + prefix_len;
  479. faux_list_add(completions, faux_str_dup(compl_str));
  480. faux_str_free(l);
  481. }
  482. faux_str_free(out);
  483. }
  484. // Put completion list to message
  485. citer = faux_list_head(completions);
  486. while ((compl_str = faux_list_each(&citer))) {
  487. faux_msg_add_param(ack, KTP_PARAM_LINE,
  488. compl_str, strlen(compl_str));
  489. }
  490. faux_list_free(completions);
  491. }
  492. faux_msg_send_async(ack, ktpd->async);
  493. faux_msg_free(ack);
  494. kpargv_free(pargv);
  495. */
  496. return BOOL_TRUE;
  497. }
  498. static bool_t ktpd_session_dispatch(ktpd_session_t *ktpd, faux_msg_t *msg)
  499. {
  500. uint16_t cmd = 0;
  501. assert(ktpd);
  502. if (!ktpd)
  503. return BOOL_FALSE;
  504. assert(msg);
  505. if (!msg)
  506. return BOOL_FALSE;
  507. cmd = faux_msg_get_cmd(msg);
  508. switch (cmd) {
  509. case KTP_CMD:
  510. ktpd_session_process_cmd(ktpd, msg);
  511. break;
  512. case KTP_COMPLETION:
  513. ktpd_session_process_completion(ktpd, msg);
  514. break;
  515. case KTP_HELP:
  516. ktpd_session_process_help(ktpd, msg);
  517. break;
  518. default:
  519. syslog(LOG_WARNING, "Unsupported command: 0x%04u\n", cmd);
  520. break;
  521. }
  522. return BOOL_TRUE;
  523. }
  524. /** @brief Low-level function to receive KTP message.
  525. *
  526. * Firstly function gets the header of message. Then it checks and parses
  527. * header and find out the length of whole message. Then it receives the rest
  528. * of message.
  529. */
  530. static bool_t ktpd_session_read_cb(faux_async_t *async,
  531. faux_buf_t *buf, size_t len, void *user_data)
  532. {
  533. ktpd_session_t *ktpd = (ktpd_session_t *)user_data;
  534. faux_msg_t *completed_msg = NULL;
  535. char *data = NULL;
  536. assert(async);
  537. assert(buf);
  538. assert(ktpd);
  539. // Linearize buffer
  540. data = malloc(len);
  541. faux_buf_read(buf, data, len);
  542. // Receive header
  543. if (!ktpd->hdr) {
  544. size_t whole_len = 0;
  545. size_t msg_wo_hdr = 0;
  546. ktpd->hdr = (faux_hdr_t *)data;
  547. // Check for broken header
  548. if (!ktp_check_header(ktpd->hdr)) {
  549. faux_free(ktpd->hdr);
  550. ktpd->hdr = NULL;
  551. return BOOL_FALSE;
  552. }
  553. whole_len = faux_hdr_len(ktpd->hdr);
  554. // msg_wo_hdr >= 0 because ktp_check_header() validates whole_len
  555. msg_wo_hdr = whole_len - sizeof(faux_hdr_t);
  556. // Plan to receive message body
  557. if (msg_wo_hdr > 0) {
  558. faux_async_set_read_limits(async,
  559. msg_wo_hdr, msg_wo_hdr);
  560. return BOOL_TRUE;
  561. }
  562. // Here message is completed (msg body has zero length)
  563. completed_msg = faux_msg_deserialize_parts(ktpd->hdr, NULL, 0);
  564. // Receive message body
  565. } else {
  566. completed_msg = faux_msg_deserialize_parts(ktpd->hdr, data, len);
  567. faux_free(data);
  568. }
  569. // Plan to receive msg header
  570. faux_async_set_read_limits(ktpd->async,
  571. sizeof(faux_hdr_t), sizeof(faux_hdr_t));
  572. faux_free(ktpd->hdr);
  573. ktpd->hdr = NULL; // Ready to recv new header
  574. // Here message is completed
  575. ktpd_session_dispatch(ktpd, completed_msg);
  576. faux_msg_free(completed_msg);
  577. return BOOL_TRUE;
  578. }
  579. bool_t ktpd_session_connected(ktpd_session_t *ktpd)
  580. {
  581. assert(ktpd);
  582. if (!ktpd)
  583. return BOOL_FALSE;
  584. if (KTPD_SESSION_STATE_DISCONNECTED == ktpd->state)
  585. return BOOL_FALSE;
  586. return BOOL_TRUE;
  587. }
  588. int ktpd_session_fd(const ktpd_session_t *ktpd)
  589. {
  590. assert(ktpd);
  591. if (!ktpd)
  592. return BOOL_FALSE;
  593. return faux_async_fd(ktpd->async);
  594. }
  595. static bool_t action_stdout_ev(faux_eloop_t *eloop, faux_eloop_type_e type,
  596. void *associated_data, void *user_data)
  597. {
  598. faux_eloop_info_fd_t *info = (faux_eloop_info_fd_t *)associated_data;
  599. ktpd_session_t *ktpd = (ktpd_session_t *)user_data;
  600. ssize_t r = -1;
  601. faux_buf_t *faux_buf = NULL;
  602. char *buf = NULL;
  603. ssize_t len = 0;
  604. faux_msg_t *ack = NULL;
  605. // Some errors or fd is closed so remove it from polling
  606. if (!(info->revents & POLLIN)) {
  607. faux_eloop_del_fd(eloop, info->fd);
  608. return BOOL_TRUE;
  609. }
  610. if (!ktpd)
  611. return BOOL_TRUE;
  612. if (!ktpd->exec)
  613. return BOOL_TRUE;
  614. faux_buf = kexec_bufout(ktpd->exec);
  615. assert(faux_buf);
  616. do {
  617. void *linear_buf = NULL;
  618. ssize_t really_readed = 0;
  619. ssize_t linear_len =
  620. faux_buf_dwrite_lock_easy(faux_buf, &linear_buf);
  621. // Non-blocked read. The fd became non-blocked while
  622. // kexec_prepare().
  623. r = read(info->fd, linear_buf, linear_len);
  624. if (r > 0)
  625. really_readed = r;
  626. faux_buf_dwrite_unlock_easy(faux_buf, really_readed);
  627. } while (r > 0);
  628. len = faux_buf_len(faux_buf);
  629. if (0 == len)
  630. return BOOL_TRUE;
  631. buf = malloc(len);
  632. faux_buf_read(faux_buf, buf, len);
  633. // Create KTP_STDOUT message to send to client
  634. ack = ktp_msg_preform(KTP_STDOUT, KTP_STATUS_NONE);
  635. faux_msg_add_param(ack, KTP_PARAM_LINE, buf, len);
  636. faux_msg_send_async(ack, ktpd->async);
  637. faux_msg_free(ack);
  638. free(buf);
  639. // Happy compiler
  640. eloop = eloop;
  641. type = type;
  642. return BOOL_TRUE;
  643. }
  644. static bool_t action_stderr_ev(faux_eloop_t *eloop, faux_eloop_type_e type,
  645. void *associated_data, void *user_data)
  646. {
  647. faux_eloop_info_fd_t *info = (faux_eloop_info_fd_t *)associated_data;
  648. ktpd_session_t *ktpd = (ktpd_session_t *)user_data;
  649. ssize_t r = -1;
  650. faux_buf_t *faux_buf = NULL;
  651. char *buf = NULL;
  652. ssize_t len = 0;
  653. faux_msg_t *ack = NULL;
  654. // Some errors or fd is closed so remove it from polling
  655. if (!(info->revents & POLLIN)) {
  656. faux_eloop_del_fd(eloop, info->fd);
  657. return BOOL_TRUE;
  658. }
  659. if (!ktpd)
  660. return BOOL_TRUE;
  661. if (!ktpd->exec)
  662. return BOOL_TRUE;
  663. faux_buf = kexec_buferr(ktpd->exec);
  664. assert(faux_buf);
  665. do {
  666. void *linear_buf = NULL;
  667. ssize_t really_readed = 0;
  668. ssize_t linear_len =
  669. faux_buf_dwrite_lock_easy(faux_buf, &linear_buf);
  670. // Non-blocked read. The fd became non-blocked while
  671. // kexec_prepare().
  672. r = read(info->fd, linear_buf, linear_len);
  673. if (r > 0)
  674. really_readed = r;
  675. faux_buf_dwrite_unlock_easy(faux_buf, really_readed);
  676. } while (r > 0);
  677. len = faux_buf_len(faux_buf);
  678. if (0 == len)
  679. return BOOL_TRUE;
  680. buf = malloc(len);
  681. faux_buf_read(faux_buf, buf, len);
  682. // Create KTP_STDERR message to send to client
  683. ack = ktp_msg_preform(KTP_STDERR, KTP_STATUS_NONE);
  684. faux_msg_add_param(ack, KTP_PARAM_LINE, buf, len);
  685. faux_msg_send_async(ack, ktpd->async);
  686. faux_msg_free(ack);
  687. free(buf);
  688. // Happy compiler
  689. eloop = eloop;
  690. type = type;
  691. return BOOL_TRUE;
  692. }
  693. bool_t client_ev(faux_eloop_t *eloop, faux_eloop_type_e type,
  694. void *associated_data, void *user_data)
  695. {
  696. faux_eloop_info_fd_t *info = (faux_eloop_info_fd_t *)associated_data;
  697. ktpd_session_t *ktpd = (ktpd_session_t *)user_data;
  698. faux_async_t *async = ktpd->async;
  699. assert(async);
  700. // Write data
  701. if (info->revents & POLLOUT) {
  702. faux_eloop_exclude_fd_event(eloop, info->fd, POLLOUT);
  703. if (faux_async_out(async) < 0) {
  704. // Someting went wrong
  705. faux_eloop_del_fd(eloop, info->fd);
  706. syslog(LOG_ERR, "Problem with async output");
  707. return BOOL_FALSE; // Stop event loop
  708. }
  709. }
  710. // Read data
  711. if (info->revents & POLLIN) {
  712. if (faux_async_in(async) < 0) {
  713. // Someting went wrong
  714. faux_eloop_del_fd(eloop, info->fd);
  715. syslog(LOG_ERR, "Problem with async input");
  716. return BOOL_FALSE; // Stop event loop
  717. }
  718. }
  719. // EOF
  720. if (info->revents & POLLHUP) {
  721. faux_eloop_del_fd(eloop, info->fd);
  722. syslog(LOG_DEBUG, "Close connection %d", info->fd);
  723. return BOOL_FALSE; // Stop event loop
  724. }
  725. // POLLERR
  726. if (info->revents & POLLERR) {
  727. faux_eloop_del_fd(eloop, info->fd);
  728. syslog(LOG_DEBUG, "POLLERR received %d", info->fd);
  729. return BOOL_FALSE; // Stop event loop
  730. }
  731. // POLLNVAL
  732. if (info->revents & POLLNVAL) {
  733. faux_eloop_del_fd(eloop, info->fd);
  734. syslog(LOG_DEBUG, "POLLNVAL received %d", info->fd);
  735. return BOOL_FALSE; // Stop event loop
  736. }
  737. type = type; // Happy compiler
  738. // Session can be really finished here. Note KTPD session can't be
  739. // stopped immediately so it's only two places within code to really
  740. // break the loop. This one and within wait_for_action_ev().
  741. if (ktpd->exit)
  742. return BOOL_FALSE;
  743. return BOOL_TRUE;
  744. }
  745. #if 0
  746. static void ktpd_session_bad_socket(ktpd_session_t *ktpd)
  747. {
  748. assert(ktpd);
  749. if (!ktpd)
  750. return;
  751. ktpd->state = KTPD_SESSION_STATE_DISCONNECTED;
  752. }
  753. #endif