hs_pirprocess.c 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435
  1. /**
  2. * \file hs_pirprocess.c
  3. * \brief Handle hidden service PIR helper processes.
  4. **/
  5. #include "core/or/or.h"
  6. #include "feature/hs/hs_pirprocess.h"
  7. #define SUBPROCESS_PRIVATE
  8. #include "lib/process/env.h"
  9. #include "lib/process/subprocess.h"
  10. #include "lib/evloop/compat_libevent.h"
  11. #include "feature/dircommon/dir_connection_st.h"
  12. #include <event2/event.h>
  13. #ifdef HAVE_UNISTD_H
  14. #include <unistd.h>
  15. #endif
  16. typedef enum {
  17. PIRPROCESS_READSTATE_HEADER,
  18. PIRPROCESS_READSTATE_BODY
  19. } PIRProcessReadState;
  20. struct pir_process_st {
  21. process_handle_t *process;
  22. buf_t *stdin_buf;
  23. buf_t *stderr_buf;
  24. struct event *stdin_ev;
  25. struct event *stdout_ev;
  26. struct event *stderr_ev;
  27. char *loglabel;
  28. pir_process_msghandler_t msghandler;
  29. PIRProcessReadState readstate;
  30. size_t readoff, readleft;
  31. unsigned char hdrbuf[PIRPROCESS_HDR_SIZE];
  32. char *bodybuf;
  33. };
  34. /* This is called when the pir process has output for us. */
  35. static void
  36. pirprocess_stdoutcb(evutil_socket_t fd, short what, void *arg)
  37. {
  38. pir_process_t *handlep = (pir_process_t*)arg;
  39. pir_process_t handle = *handlep;
  40. if (!(what & EV_READ)) {
  41. /* Not sure why we're here */
  42. return;
  43. }
  44. if (handle->readstate == PIRPROCESS_READSTATE_HEADER) {
  45. int res = read(fd, handle->hdrbuf + handle->readoff, handle->readleft);
  46. if (res <= 0) {
  47. /* Tell the caller that the PIR process failed to send us any output. */
  48. tor_free(handle->bodybuf);
  49. handle->readoff = 0;
  50. handle->readleft = PIRPROCESS_HDR_SIZE;
  51. handle->readstate = PIRPROCESS_READSTATE_HEADER;
  52. (handle->msghandler)(NULL, NULL, 0);
  53. hs_pirprocess_close(handlep);
  54. return;
  55. }
  56. handle->readoff += res;
  57. handle->readleft -= res;
  58. if (handle->readleft == 0) {
  59. handle->readleft = ntohl(*(uint32_t*)
  60. (handle->hdrbuf+PIRPROCESS_HDR_SIZE-4));
  61. tor_free(handle->bodybuf);
  62. if (handle->readleft > 0) {
  63. handle->bodybuf = tor_malloc(handle->readleft+1);
  64. handle->bodybuf[handle->readleft] = '\0';
  65. handle->readoff = 0;
  66. handle->readstate = PIRPROCESS_READSTATE_BODY;
  67. } else {
  68. (handle->msghandler)(handle->hdrbuf, NULL, 0);
  69. handle->readoff = 0;
  70. handle->readleft = PIRPROCESS_HDR_SIZE;
  71. handle->readstate = PIRPROCESS_READSTATE_HEADER;
  72. }
  73. }
  74. } else if (handle->readstate == PIRPROCESS_READSTATE_BODY) {
  75. int res = read(fd, handle->bodybuf + handle->readoff,
  76. handle->readleft);
  77. if (res <= 0) {
  78. /* Tell the caller that the PIR process failed to send us any output. */
  79. tor_free(handle->bodybuf);
  80. handle->readoff = 0;
  81. handle->readleft = PIRPROCESS_HDR_SIZE;
  82. handle->readstate = PIRPROCESS_READSTATE_HEADER;
  83. (handle->msghandler)(NULL, NULL, 0);
  84. hs_pirprocess_close(handlep);
  85. return;
  86. }
  87. handle->readoff += res;
  88. handle->readleft -= res;
  89. if (handle->readleft == 0) {
  90. /* Reading is complete */
  91. (handle->msghandler)(handle->hdrbuf, handle->bodybuf,
  92. handle->readoff);
  93. /* Prepare for the next output from the PIR server */
  94. tor_free(handle->bodybuf);
  95. handle->readoff = 0;
  96. handle->readleft = PIRPROCESS_HDR_SIZE;
  97. handle->readstate = PIRPROCESS_READSTATE_HEADER;
  98. }
  99. }
  100. }
  101. /* This is called when the pir process is ready to read from its stdin. */
  102. static void
  103. pirprocess_stdincb(evutil_socket_t fd, short what, void *arg)
  104. {
  105. pir_process_t *handlep = (pir_process_t*)arg;
  106. pir_process_t handle = *handlep;
  107. int res;
  108. size_t bufsize = buf_datalen(handle->stdin_buf);
  109. char *netbuf = NULL;
  110. if (!(what & EV_WRITE)) {
  111. /* Not sure why we're here */
  112. log_info(LD_DIRSERV,"PIRSERVER bailing");
  113. return;
  114. }
  115. if (bufsize == 0) {
  116. log_err(LD_DIRSERV,"PIRSERVER trying to write 0-length buffer");
  117. return;
  118. }
  119. netbuf = tor_malloc(bufsize);
  120. if (netbuf == NULL) {
  121. log_err(LD_DIRSERV,"PIRSERVER failed to allocate buffer");
  122. return;
  123. }
  124. /* One might think that just calling buf_flush_to_socket would be
  125. * the thing to do, but that function ends up calling sendto()
  126. * instead of write(), which doesn't work on pipes. So we do it
  127. * more manually. Using a bufferevent may be another option. */
  128. buf_peek(handle->stdin_buf, netbuf, bufsize);
  129. res = write(fd, netbuf, bufsize);
  130. tor_free(netbuf);
  131. if (res < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) {
  132. /* Try writing again later */
  133. event_add(handle->stdin_ev, NULL);
  134. return;
  135. }
  136. if (res <= 0) {
  137. /* Stop trying to write. */
  138. return;
  139. }
  140. buf_drain(handle->stdin_buf, res);
  141. bufsize -= res;
  142. if (bufsize > 0) {
  143. /* There's more to write */
  144. event_add(handle->stdin_ev, NULL);
  145. }
  146. }
  147. /* This is called when the pir process writes something to its stderr. */
  148. static void
  149. pirprocess_stderrcb(evutil_socket_t fd, short what, void *arg)
  150. {
  151. pir_process_t *handlep = (pir_process_t*)arg;
  152. pir_process_t handle = *handlep;
  153. if (!(what & EV_READ)) {
  154. /* Not sure why we're here */
  155. return;
  156. }
  157. /* Add the data to the stderr buffer */
  158. char buf[1000];
  159. int res = read(fd, buf, sizeof(buf)-1);
  160. if (res <= 0) return;
  161. buf[res] = '\0';
  162. buf_add(handle->stderr_buf, buf, res);
  163. /* Read lines from the stderr_buf */
  164. while (1) {
  165. size_t linesize = 0;
  166. res = buf_get_line(handle->stderr_buf, NULL, &linesize);
  167. if (res == 0) {
  168. /* No complete lines in the buffer */
  169. break;
  170. }
  171. char *linebuf = tor_malloc(linesize);
  172. buf_get_line(handle->stderr_buf, linebuf, &linesize);
  173. if (linesize > 0 && linebuf[linesize-1] == '\n') {
  174. linebuf[linesize-1] = '\0';
  175. }
  176. log_info(LD_DIRSERV,"%s %s", handle->loglabel, escaped(linebuf));
  177. tor_free(linebuf);
  178. }
  179. }
  180. void
  181. hs_pirprocess_close(pir_process_t *handlep)
  182. {
  183. pir_process_t handle;
  184. if (!handlep) return;
  185. if (!*handlep) return;
  186. handle = *handlep;
  187. if (handle->stdin_buf) {
  188. buf_free(handle->stdin_buf);
  189. }
  190. if (handle->stderr_buf) {
  191. buf_free(handle->stderr_buf);
  192. }
  193. if (handle->stdin_ev) {
  194. event_free(handle->stdin_ev);
  195. }
  196. if (handle->stdout_ev) {
  197. event_free(handle->stdout_ev);
  198. }
  199. if (handle->stderr_ev) {
  200. event_free(handle->stderr_ev);
  201. }
  202. if (handle->process) {
  203. log_info(LD_DIRSERV,"PIRPROCESS destroying subprocess");
  204. tor_process_handle_destroy(handle->process, 1);
  205. }
  206. if (handle->loglabel) {
  207. tor_free(handle->loglabel);
  208. }
  209. if (handle->bodybuf) {
  210. tor_free(handle->bodybuf);
  211. }
  212. tor_free(handle);
  213. *handlep = NULL;
  214. }
  215. void
  216. hs_pirprocess_poke(const char *path, const char *loglabel,
  217. pir_process_msghandler_t msghandler, pir_process_t *handlep)
  218. {
  219. int res;
  220. smartlist_t *env_vars = get_current_process_environment_variables();
  221. const char *argv[2];
  222. process_environment_t *env;
  223. pir_process_t handle;
  224. if (!handlep) return;
  225. if (*handlep) {
  226. handle = *handlep;
  227. if (handle->process &&
  228. handle->process->status == PROCESS_STATUS_RUNNING) {
  229. /* The PIR process appears to be open */
  230. return;
  231. }
  232. }
  233. /* Close everything we've got and start over */
  234. hs_pirprocess_close(handlep);
  235. if (!path) {
  236. /* We don't have a configured PIR server */
  237. return;
  238. }
  239. *handlep = tor_malloc_zero(sizeof(struct pir_process_st));
  240. handle = *handlep;
  241. /* Start the process */
  242. argv[0] = path;
  243. argv[1] = NULL;
  244. env = process_environment_make(env_vars);
  245. log_info(LD_DIRSERV,"PIRPROCESS spawning subprocess");
  246. res = tor_spawn_background(path, argv, env, &(handle->process));
  247. SMARTLIST_FOREACH(env_vars, void *, x, tor_free(x));
  248. smartlist_free(env_vars);
  249. if (res != PROCESS_STATUS_RUNNING) {
  250. /* Launch failure */
  251. hs_pirprocess_close(handlep);
  252. return;
  253. }
  254. handle->loglabel = tor_strdup(loglabel);
  255. handle->msghandler = msghandler;
  256. /* Create a libevent event to listen to the PIR process' responses. */
  257. handle->stdout_ev = event_new(tor_libevent_get_base(),
  258. handle->process->stdout_pipe, EV_READ|EV_PERSIST,
  259. pirprocess_stdoutcb, handlep);
  260. event_add(handle->stdout_ev, NULL);
  261. /* And one to listen to the PIR server's stderr. */
  262. handle->stderr_ev = event_new(tor_libevent_get_base(),
  263. handle->process->stderr_pipe, EV_READ|EV_PERSIST,
  264. pirprocess_stderrcb, handlep);
  265. event_add(handle->stderr_ev, NULL);
  266. /* And one for writability to the pir process' stdin, but don't add
  267. * it just yet. Also create the buffer it will use. */
  268. handle->stdin_buf = buf_new();
  269. handle->stderr_buf = buf_new();
  270. handle->stdin_ev = event_new(tor_libevent_get_base(),
  271. handle->process->stdin_pipe, EV_WRITE, pirprocess_stdincb,
  272. handlep);
  273. handle->readstate = PIRPROCESS_READSTATE_HEADER;
  274. handle->readoff = 0;
  275. handle->readleft = PIRPROCESS_HDR_SIZE;
  276. handle->bodybuf = NULL;
  277. }
  278. int
  279. hs_pirprocess_send(pir_process_t handle, const unsigned char *buf,
  280. size_t len)
  281. {
  282. if (handle == NULL || handle->process == NULL ||
  283. handle->process->status != PROCESS_STATUS_RUNNING) {
  284. /* We don't have a process to talk to */
  285. return -1;
  286. }
  287. /* Write the data to the stdin buffer */
  288. if (len > 0) {
  289. buf_add(handle->stdin_buf, (const char *)buf, len);
  290. event_add(handle->stdin_ev, NULL);
  291. }
  292. return len;
  293. }
  294. typedef struct {
  295. dir_connection_t *conn;
  296. pir_process_abort_fn abort_fn;
  297. } pirprocess_digestmap_entry_t;
  298. /* What we really want is a map from uint64_t to
  299. * pirprocess_digestmap_entry_t*, but we've got an implementation of
  300. * char[20] to void*, which we'll repurpose for this use. Note that the
  301. * dir_connection_t* pointers in this map are *not* owned by this map,
  302. * and this map must not free them. */
  303. static digestmap_t *pirprocess_reqid_map;
  304. /* The last-used request id */
  305. static uint64_t pirprocess_last_reqid;
  306. void
  307. hs_pirprocess_init(void)
  308. {
  309. pirprocess_reqid_map = digestmap_new();
  310. pirprocess_last_reqid = 0;
  311. }
  312. static void
  313. entry_abort_free(void *p)
  314. {
  315. pirprocess_digestmap_entry_t *entry = (pirprocess_digestmap_entry_t *)p;
  316. if (entry->abort_fn) {
  317. (entry->abort_fn)(entry->conn);
  318. }
  319. tor_free(entry);
  320. }
  321. void
  322. hs_pirprocess_abort_all(void)
  323. {
  324. digestmap_free(pirprocess_reqid_map, entry_abort_free);
  325. hs_pirprocess_init();
  326. }
  327. void
  328. hs_pirprocess_free_all(void)
  329. {
  330. digestmap_free(pirprocess_reqid_map, tor_free_);
  331. }
  332. uint64_t
  333. hs_pirprocess_alloc_reqid(dir_connection_t *dir_conn,
  334. pir_process_abort_fn abort_fn)
  335. {
  336. char mapbuf[DIGEST_LEN];
  337. pirprocess_digestmap_entry_t *entry;
  338. uint64_t reqid = dir_conn->pirprocess_reqid;
  339. if (reqid > 0) {
  340. return reqid;
  341. }
  342. ++pirprocess_last_reqid;
  343. if (pirprocess_last_reqid == 0) {
  344. /* We wrapped a 64-bit integer?! */
  345. pirprocess_last_reqid = 1;
  346. }
  347. reqid = pirprocess_last_reqid;
  348. memset(mapbuf, 0, DIGEST_LEN);
  349. memmove(mapbuf, &reqid, sizeof(reqid));
  350. entry = tor_malloc(sizeof(*entry));
  351. entry->conn = dir_conn;
  352. entry->abort_fn = abort_fn;
  353. digestmap_set(pirprocess_reqid_map, mapbuf, entry);
  354. dir_conn->pirprocess_reqid = reqid;
  355. return dir_conn->pirprocess_reqid;
  356. }
  357. void
  358. hs_pirprocess_dealloc_reqid(dir_connection_t *dir_conn)
  359. {
  360. char mapbuf[DIGEST_LEN];
  361. uint64_t reqid = dir_conn->pirprocess_reqid;
  362. pirprocess_digestmap_entry_t *entry;
  363. if (reqid == 0) return;
  364. memset(mapbuf, 0, DIGEST_LEN);
  365. memmove(mapbuf, &reqid, sizeof(reqid));
  366. entry = digestmap_remove(pirprocess_reqid_map, mapbuf);
  367. tor_free(entry);
  368. dir_conn->pirprocess_reqid = 0;
  369. }
  370. dir_connection_t *
  371. hs_pirprocess_lookup_reqid(uint64_t reqid)
  372. {
  373. char mapbuf[DIGEST_LEN];
  374. pirprocess_digestmap_entry_t *entry;
  375. memset(mapbuf, 0, DIGEST_LEN);
  376. memmove(mapbuf, &reqid, sizeof(reqid));
  377. entry = digestmap_get(pirprocess_reqid_map, mapbuf);
  378. return entry ? entry->conn : NULL;
  379. }