RPC.c 7.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267
  1. #include <stdio.h>
  2. #include <unistd.h>
  3. #include <sys/un.h>
  4. #include <pthread.h>
  5. #include <errno.h>
  6. #include <sys/syscall.h>
  7. #include <fcntl.h>
  8. #include <stdint.h>
  9. #include <sys/socket.h>
  10. #include <strings.h>
  11. #include "RPC.h"
  12. #define RPC_FD 1023
  13. #define SERVICE_CONNECT_ATTEMPTS 30
  14. static int instance_count;
  15. static int rpc_count;
  16. static pthread_mutex_t lock;
  17. void rpc_mutex_init() {
  18. if(pthread_mutex_init(&lock, NULL) != 0) {
  19. fprintf(stderr, "error while initializing service call mutex\n");
  20. }
  21. }
  22. void rpc_mutex_destroy() {
  23. pthread_mutex_destroy(&lock);
  24. }
  25. /*
  26. * Reads a new file descriptor from the service
  27. */
  28. int get_new_fd(int sock)
  29. {
  30. char buf[BUF_SZ];
  31. int newfd;
  32. ssize_t size = sock_fd_read(sock, buf, sizeof(buf), &newfd);
  33. if(size > 0){
  34. return newfd;
  35. }
  36. fprintf(stderr, "get_new_fd(): Error, unable to read fd over (%d)\n", sock);
  37. return -1;
  38. }
  39. /*
  40. * Reads a return value from the service and sets errno (if applicable)
  41. */
  42. int get_retval(int rpc_sock)
  43. {
  44. if(rpc_sock >= 0) {
  45. int retval;
  46. int sz = sizeof(char) + sizeof(retval) + sizeof(errno);
  47. char retbuf[BUF_SZ];
  48. memset(&retbuf, 0, sz);
  49. int n_read = read(rpc_sock, &retbuf, sz);
  50. if(n_read > 0) {
  51. memcpy(&retval, &retbuf[1], sizeof(retval));
  52. memcpy(&errno, &retbuf[1+sizeof(retval)], sizeof(errno));
  53. return retval;
  54. }
  55. }
  56. return -1;
  57. }
  58. int rpc_join(const char * sockname)
  59. {
  60. struct sockaddr_un addr;
  61. int conn_err = -1, attempts = 0;
  62. memset(&addr, 0, sizeof(addr));
  63. addr.sun_family = AF_UNIX;
  64. strncpy(addr.sun_path, sockname, sizeof(addr.sun_path)-1);
  65. int sock;
  66. if((sock = socket(AF_UNIX, SOCK_STREAM, 0)) < 0){
  67. fprintf(stderr, "Error while creating RPC socket\n");
  68. return -1;
  69. }
  70. while((conn_err != 0) && (attempts < SERVICE_CONNECT_ATTEMPTS)){
  71. if((conn_err = connect(sock, (struct sockaddr*)&addr, sizeof(addr))) != 0) {
  72. fprintf(stderr, "Error while connecting to RPC socket. Re-attempting...\n");
  73. sleep(1);
  74. }
  75. else {
  76. //int newfd = dup2(sock, RPC_FD-instance_count);
  77. //close(sock);
  78. return sock;
  79. }
  80. attempts++;
  81. }
  82. return -1;
  83. }
  84. /*
  85. * Send a command to the service
  86. */
  87. int rpc_send_command(int cmd, int forfd, void *data, int len)
  88. {
  89. pthread_mutex_lock(&lock);
  90. char c, padding[] = {0, 1, 1, 2, 3, 5, 8, 13, 21, 34, 55, 89};
  91. char cmdbuf[BUF_SZ], magic[TOKEN_SIZE], metabuf[BUF_SZ];
  92. memcpy(magic+MAGIC_SIZE, padding, TOKEN_SIZE);
  93. uint64_t magic_num;
  94. // ephemeral RPC socket used only for this command
  95. int rpc_sock = rpc_join("/root/dev/ztest/nc_e5cd7a9e1c3511dd");
  96. // Generate token
  97. int fdrand = open("/dev/urandom", O_RDONLY);
  98. read(fdrand, &magic, MAGIC_SIZE);
  99. memcpy(&magic_num, magic, MAGIC_SIZE);
  100. cmdbuf[CMD_ID_IDX] = cmd;
  101. memcpy(&cmdbuf[MAGIC_IDX], &magic_num, MAGIC_SIZE);
  102. memcpy(&cmdbuf[STRUCT_IDX], data, len);
  103. // Format: [sig_byte] + [cmd_id] + [magic] + [meta] + [payload]
  104. #ifdef VERBOSE
  105. /*
  106. #define IDX_PID 0
  107. #define IDX_TID sizeof(pid_t)
  108. #define IDX_COUNT IDX_TID + sizeof(pid_t)
  109. #define IDX_TIME IDX_COUNT + sizeof(int)
  110. #define IDX_CMD IDX_TIME + 20 // 20 being the length of the timestamp string
  111. #define IDX_PAYLOAD IDX_TIME + sizeof(char)
  112. */
  113. /* [pid_t] [pid_t] [rpc_count] [int] [...] */
  114. memset(metabuf, 0, BUF_SZ);
  115. pid_t pid = syscall(SYS_getpid);
  116. pid_t tid = syscall(SYS_gettid);
  117. rpc_count++;
  118. char timestring[20];
  119. time_t timestamp;
  120. timestamp = time(NULL);
  121. strftime(timestring, sizeof(timestring), "%H:%M:%S", localtime(&timestamp));
  122. metabuf[IDX_SIGNAL_BYTE] = 'R';
  123. memcpy(&metabuf[IDX_PID], &pid, sizeof(pid_t) ); /* pid */
  124. memcpy(&metabuf[IDX_TID], &tid, sizeof(pid_t) ); /* tid */
  125. memcpy(&metabuf[IDX_COUNT], &rpc_count, sizeof(rpc_count) ); /* rpc_count */
  126. memcpy(&metabuf[IDX_TIME], &timestring, 20 ); /* timestamp */
  127. #endif
  128. /* Combine command flag+payload with RPC metadata */
  129. memcpy(&metabuf[IDX_PAYLOAD], cmdbuf, len + 1 + MAGIC_SIZE);
  130. // Write RPC
  131. int n_write = write(rpc_sock, &metabuf, BUF_SZ);
  132. if(n_write < 0) {
  133. fprintf(stderr, "Error writing command to service (CMD = %d)\n", cmdbuf[CMD_ID_IDX]);
  134. errno = 0;
  135. }
  136. // Write token to corresponding data stream
  137. read(rpc_sock, &c, 1);
  138. if(c == 'z' && n_write > 0 && forfd > -1){
  139. int w = send(forfd, &magic, TOKEN_SIZE, 0);
  140. }
  141. // Process response from service
  142. int ret = ERR_OK;
  143. if(n_write > 0) {
  144. if(cmdbuf[CMD_ID_IDX]==RPC_SOCKET) {
  145. pthread_mutex_unlock(&lock);
  146. return rpc_sock; // Used as new socket
  147. }
  148. if(cmdbuf[CMD_ID_IDX]==RPC_CONNECT
  149. || cmdbuf[CMD_ID_IDX]==RPC_BIND
  150. || cmdbuf[CMD_ID_IDX]==RPC_LISTEN) {
  151. ret = get_retval(rpc_sock);
  152. }
  153. if(cmdbuf[CMD_ID_IDX]==RPC_GETSOCKNAME) {
  154. pthread_mutex_unlock(&lock);
  155. return rpc_sock; // Don't close rpc here, we'll use it to read getsockopt_st
  156. }
  157. }
  158. else
  159. ret = -1;
  160. close(rpc_sock); // We're done with this RPC socket, close it (if type-R)
  161. pthread_mutex_unlock(&lock);
  162. return ret;
  163. }
  164. /*
  165. * Send file descriptor
  166. */
  167. ssize_t sock_fd_write(int sock, int fd)
  168. {
  169. ssize_t size;
  170. struct msghdr msg;
  171. struct iovec iov;
  172. char buf = '\0';
  173. int buflen = 1;
  174. union {
  175. struct cmsghdr cmsghdr;
  176. char control[CMSG_SPACE(sizeof (int))];
  177. } cmsgu;
  178. struct cmsghdr *cmsg;
  179. iov.iov_base = &buf;
  180. iov.iov_len = buflen;
  181. msg.msg_name = NULL;
  182. msg.msg_namelen = 0;
  183. msg.msg_iov = &iov;
  184. msg.msg_iovlen = 1;
  185. if (fd != -1) {
  186. msg.msg_control = cmsgu.control;
  187. msg.msg_controllen = sizeof(cmsgu.control);
  188. cmsg = CMSG_FIRSTHDR(&msg);
  189. cmsg->cmsg_len = CMSG_LEN(sizeof (int));
  190. cmsg->cmsg_level = SOL_SOCKET;
  191. cmsg->cmsg_type = SCM_RIGHTS;
  192. *((int *) CMSG_DATA(cmsg)) = fd;
  193. } else {
  194. msg.msg_control = NULL;
  195. msg.msg_controllen = 0;
  196. }
  197. size = sendmsg(sock, &msg, 0);
  198. if (size < 0)
  199. perror ("sendmsg");
  200. return size;
  201. }
  202. /*
  203. * Read a file descriptor
  204. */
  205. ssize_t sock_fd_read(int sock, void *buf, ssize_t bufsize, int *fd)
  206. {
  207. ssize_t size;
  208. if (fd) {
  209. struct msghdr msg;
  210. struct iovec iov;
  211. union {
  212. struct cmsghdr cmsghdr;
  213. char control[CMSG_SPACE(sizeof (int))];
  214. } cmsgu;
  215. struct cmsghdr *cmsg;
  216. iov.iov_base = buf;
  217. iov.iov_len = bufsize;
  218. msg.msg_name = NULL;
  219. msg.msg_namelen = 0;
  220. msg.msg_iov = &iov;
  221. msg.msg_iovlen = 1;
  222. msg.msg_control = cmsgu.control;
  223. msg.msg_controllen = sizeof(cmsgu.control);
  224. size = recvmsg (sock, &msg, 0);
  225. if (size < 0) {
  226. fprintf(stderr, "sock_fd_read(): recvmsg: Error\n");
  227. return -1;
  228. }
  229. cmsg = CMSG_FIRSTHDR(&msg);
  230. if (cmsg && cmsg->cmsg_len == CMSG_LEN(sizeof(int))) {
  231. if (cmsg->cmsg_level != SOL_SOCKET) {
  232. fprintf (stderr, "invalid cmsg_level %d\n",cmsg->cmsg_level);
  233. return -1;
  234. }
  235. if (cmsg->cmsg_type != SCM_RIGHTS) {
  236. fprintf (stderr, "invalid cmsg_type %d\n",cmsg->cmsg_type);
  237. return -1;
  238. }
  239. *fd = *((int *) CMSG_DATA(cmsg));
  240. } else *fd = -1;
  241. } else {
  242. size = read (sock, buf, bufsize);
  243. if (size < 0) {
  244. fprintf(stderr, "sock_fd_read(): read: Error\n");
  245. return -1;
  246. }
  247. }
  248. return size;
  249. }