#include #include #include #include #include #include #include #include #include #include #include "RPC.h" #define RPC_FD 1023 #define SERVICE_CONNECT_ATTEMPTS 30 static int instance_count; static int rpc_count; static pthread_mutex_t lock; void rpc_mutex_init() { if(pthread_mutex_init(&lock, NULL) != 0) { fprintf(stderr, "error while initializing service call mutex\n"); } } void rpc_mutex_destroy() { pthread_mutex_destroy(&lock); } /* * Reads a new file descriptor from the service */ int get_new_fd(int sock) { char buf[BUF_SZ]; int newfd; ssize_t size = sock_fd_read(sock, buf, sizeof(buf), &newfd); if(size > 0){ return newfd; } fprintf(stderr, "get_new_fd(): Error, unable to read fd over (%d)\n", sock); return -1; } /* * Reads a return value from the service and sets errno (if applicable) */ int get_retval(int rpc_sock) { if(rpc_sock >= 0) { int retval; int sz = sizeof(char) + sizeof(retval) + sizeof(errno); char retbuf[BUF_SZ]; memset(&retbuf, 0, sz); int n_read = read(rpc_sock, &retbuf, sz); if(n_read > 0) { memcpy(&retval, &retbuf[1], sizeof(retval)); memcpy(&errno, &retbuf[1+sizeof(retval)], sizeof(errno)); return retval; } } return -1; } int rpc_join(const char * sockname) { struct sockaddr_un addr; int conn_err = -1, attempts = 0; memset(&addr, 0, sizeof(addr)); addr.sun_family = AF_UNIX; strncpy(addr.sun_path, sockname, sizeof(addr.sun_path)-1); int sock; if((sock = socket(AF_UNIX, SOCK_STREAM, 0)) < 0){ fprintf(stderr, "Error while creating RPC socket\n"); return -1; } while((conn_err != 0) && (attempts < SERVICE_CONNECT_ATTEMPTS)){ if((conn_err = connect(sock, (struct sockaddr*)&addr, sizeof(addr))) != 0) { fprintf(stderr, "Error while connecting to RPC socket. Re-attempting...\n"); sleep(1); } else { //int newfd = dup2(sock, RPC_FD-instance_count); //close(sock); return sock; } attempts++; } return -1; } /* * Send a command to the service */ int rpc_send_command(int cmd, int forfd, void *data, int len) { pthread_mutex_lock(&lock); char c, padding[] = {0, 1, 1, 2, 3, 5, 8, 13, 21, 34, 55, 89}; char cmdbuf[BUF_SZ], magic[TOKEN_SIZE], metabuf[BUF_SZ]; memcpy(magic+MAGIC_SIZE, padding, TOKEN_SIZE); uint64_t magic_num; // ephemeral RPC socket used only for this command int rpc_sock = rpc_join("/root/dev/ztest/nc_e5cd7a9e1c3511dd"); // Generate token int fdrand = open("/dev/urandom", O_RDONLY); read(fdrand, &magic, MAGIC_SIZE); memcpy(&magic_num, magic, MAGIC_SIZE); cmdbuf[CMD_ID_IDX] = cmd; memcpy(&cmdbuf[MAGIC_IDX], &magic_num, MAGIC_SIZE); memcpy(&cmdbuf[STRUCT_IDX], data, len); // Format: [sig_byte] + [cmd_id] + [magic] + [meta] + [payload] #ifdef VERBOSE /* #define IDX_PID 0 #define IDX_TID sizeof(pid_t) #define IDX_COUNT IDX_TID + sizeof(pid_t) #define IDX_TIME IDX_COUNT + sizeof(int) #define IDX_CMD IDX_TIME + 20 // 20 being the length of the timestamp string #define IDX_PAYLOAD IDX_TIME + sizeof(char) */ /* [pid_t] [pid_t] [rpc_count] [int] [...] */ memset(metabuf, 0, BUF_SZ); pid_t pid = syscall(SYS_getpid); pid_t tid = syscall(SYS_gettid); rpc_count++; char timestring[20]; time_t timestamp; timestamp = time(NULL); strftime(timestring, sizeof(timestring), "%H:%M:%S", localtime(×tamp)); memcpy(metabuf, RPC_PHRASE, RPC_PHRASE_SIZE); // Write signal phrase memcpy(&metabuf[IDX_PID], &pid, sizeof(pid_t) ); /* pid */ memcpy(&metabuf[IDX_TID], &tid, sizeof(pid_t) ); /* tid */ memcpy(&metabuf[IDX_COUNT], &rpc_count, sizeof(rpc_count) ); /* rpc_count */ memcpy(&metabuf[IDX_TIME], ×tring, 20 ); /* timestamp */ #endif /* Combine command flag+payload with RPC metadata */ memcpy(&metabuf[IDX_PAYLOAD], cmdbuf, len + 1 + MAGIC_SIZE); // Write RPC int n_write = write(rpc_sock, &metabuf, BUF_SZ); if(n_write < 0) { fprintf(stderr, "Error writing command to service (CMD = %d)\n", cmdbuf[CMD_ID_IDX]); errno = 0; } // Write token to corresponding data stream read(rpc_sock, &c, 1); if(c == 'z' && n_write > 0 && forfd > -1){ int w = send(forfd, &magic, TOKEN_SIZE, 0); } // Process response from service int ret = ERR_OK; if(n_write > 0) { if(cmdbuf[CMD_ID_IDX]==RPC_SOCKET) { pthread_mutex_unlock(&lock); return rpc_sock; // Used as new socket } if(cmdbuf[CMD_ID_IDX]==RPC_CONNECT || cmdbuf[CMD_ID_IDX]==RPC_BIND || cmdbuf[CMD_ID_IDX]==RPC_LISTEN) { ret = get_retval(rpc_sock); } if(cmdbuf[CMD_ID_IDX]==RPC_GETSOCKNAME) { pthread_mutex_unlock(&lock); return rpc_sock; // Don't close rpc here, we'll use it to read getsockopt_st } } else ret = -1; close(rpc_sock); // We're done with this RPC socket, close it (if type-R) pthread_mutex_unlock(&lock); return ret; } /* * Send file descriptor */ ssize_t sock_fd_write(int sock, int fd) { ssize_t size; struct msghdr msg; struct iovec iov; char buf = '\0'; int buflen = 1; union { struct cmsghdr cmsghdr; char control[CMSG_SPACE(sizeof (int))]; } cmsgu; struct cmsghdr *cmsg; iov.iov_base = &buf; iov.iov_len = buflen; msg.msg_name = NULL; msg.msg_namelen = 0; msg.msg_iov = &iov; msg.msg_iovlen = 1; if (fd != -1) { msg.msg_control = cmsgu.control; msg.msg_controllen = sizeof(cmsgu.control); cmsg = CMSG_FIRSTHDR(&msg); cmsg->cmsg_len = CMSG_LEN(sizeof (int)); cmsg->cmsg_level = SOL_SOCKET; cmsg->cmsg_type = SCM_RIGHTS; *((int *) CMSG_DATA(cmsg)) = fd; } else { msg.msg_control = NULL; msg.msg_controllen = 0; } size = sendmsg(sock, &msg, 0); if (size < 0) perror ("sendmsg"); return size; } /* * Read a file descriptor */ ssize_t sock_fd_read(int sock, void *buf, ssize_t bufsize, int *fd) { ssize_t size; if (fd) { struct msghdr msg; struct iovec iov; union { struct cmsghdr cmsghdr; char control[CMSG_SPACE(sizeof (int))]; } cmsgu; struct cmsghdr *cmsg; iov.iov_base = buf; iov.iov_len = bufsize; msg.msg_name = NULL; msg.msg_namelen = 0; msg.msg_iov = &iov; msg.msg_iovlen = 1; msg.msg_control = cmsgu.control; msg.msg_controllen = sizeof(cmsgu.control); size = recvmsg (sock, &msg, 0); if (size < 0) { fprintf(stderr, "sock_fd_read(): recvmsg: Error\n"); return -1; } cmsg = CMSG_FIRSTHDR(&msg); if (cmsg && cmsg->cmsg_len == CMSG_LEN(sizeof(int))) { if (cmsg->cmsg_level != SOL_SOCKET) { fprintf (stderr, "invalid cmsg_level %d\n",cmsg->cmsg_level); return -1; } if (cmsg->cmsg_type != SCM_RIGHTS) { fprintf (stderr, "invalid cmsg_type %d\n",cmsg->cmsg_type); return -1; } *fd = *((int *) CMSG_DATA(cmsg)); } else *fd = -1; } else { size = read (sock, buf, bufsize); if (size < 0) { fprintf(stderr, "sock_fd_read(): read: Error\n"); return -1; } } return size; }