2 * Unix SMB/CIFS implementation.
3 * Copyright (C) Volker Lendecke 2013
5 * This program is free software; you can redistribute it and/or modify
6 * it under the terms of the GNU General Public License as published by
7 * the Free Software Foundation; either version 3 of the License, or
8 * (at your option) any later version.
10 * This program is distributed in the hope that it will be useful,
11 * but WITHOUT ANY WARRANTY; without even the implied warranty of
12 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13 * GNU General Public License for more details.
15 * You should have received a copy of the GNU General Public License
16 * along with this program. If not, see <http://www.gnu.org/licenses/>.
21 #include "system/select.h"
22 #include "system/time.h"
23 #include "system/network.h"
24 #include "dlinklist.h"
25 #include "pthreadpool/pthreadpool.h"
29 * This file implements two abstractions: The "unix_dgram" functions implement
30 * queueing for unix domain datagram sockets. You can send to a destination
31 * socket, and if that has no free space available, it will fall back to an
32 * anonymous socket that will poll for writability. "unix_dgram" expects the
33 * data size not to exceed the system limit.
35 * The "unix_msg" functions implement the fragmentation of large messages on
36 * top of "unix_dgram". This is what is exposed to the user of this API.
39 struct unix_dgram_msg {
40 struct unix_dgram_msg *prev, *next;
51 struct unix_dgram_send_queue {
52 struct unix_dgram_send_queue *prev, *next;
53 struct unix_dgram_ctx *ctx;
55 struct unix_dgram_msg *msgs;
59 struct unix_dgram_ctx {
62 const struct poll_funcs *ev_funcs;
65 void (*recv_callback)(struct unix_dgram_ctx *ctx,
66 uint8_t *msg, size_t msg_len,
67 int *fds, size_t num_fds,
71 struct poll_watch *sock_read_watch;
72 struct unix_dgram_send_queue *send_queues;
74 struct pthreadpool *send_pool;
75 struct poll_watch *pool_read_watch;
81 static ssize_t iov_buflen(const struct iovec *iov, int iovlen);
82 static void unix_dgram_recv_handler(struct poll_watch *w, int fd, short events,
85 /* Set socket non blocking. */
86 static int prepare_socket_nonblock(int sock)
90 #define FLAG_TO_SET O_NONBLOCK
93 #define FLAG_TO_SET O_NDELAY
95 #define FLAG_TO_SET FNDELAY
99 flags = fcntl(sock, F_GETFL);
103 flags |= FLAG_TO_SET;
104 if (fcntl(sock, F_SETFL, flags) == -1) {
112 /* Set socket close on exec. */
113 static int prepare_socket_cloexec(int sock)
118 flags = fcntl(sock, F_GETFD, 0);
123 if (fcntl(sock, F_SETFD, flags) == -1) {
130 /* Set socket non blocking and close on exec. */
131 static int prepare_socket(int sock)
133 int ret = prepare_socket_nonblock(sock);
138 return prepare_socket_cloexec(sock);
141 static void close_fd_array(int *fds, size_t num_fds)
145 for (i = 0; i < num_fds; i++) {
155 static int unix_dgram_init(const struct sockaddr_un *addr, size_t max_msg,
156 const struct poll_funcs *ev_funcs,
157 void (*recv_callback)(struct unix_dgram_ctx *ctx,
158 uint8_t *msg, size_t msg_len,
159 int *fds, size_t num_fds,
162 struct unix_dgram_ctx **result)
164 struct unix_dgram_ctx *ctx;
169 pathlen = strlen(addr->sun_path)+1;
174 ctx = malloc(offsetof(struct unix_dgram_ctx, path) + pathlen);
179 memcpy(ctx->path, addr->sun_path, pathlen);
184 *ctx = (struct unix_dgram_ctx) {
186 .ev_funcs = ev_funcs,
187 .recv_callback = recv_callback,
188 .private_data = private_data,
189 .created_pid = (pid_t)-1
192 ctx->recv_buf = malloc(max_msg);
193 if (ctx->recv_buf == NULL) {
198 ctx->sock = socket(AF_UNIX, SOCK_DGRAM, 0);
199 if (ctx->sock == -1) {
204 /* Set non-blocking and close-on-exec. */
205 ret = prepare_socket(ctx->sock);
211 ret = bind(ctx->sock,
212 (const struct sockaddr *)(const void *)addr,
219 ctx->created_pid = getpid();
221 ctx->sock_read_watch = ctx->ev_funcs->watch_new(
222 ctx->ev_funcs, ctx->sock, POLLIN,
223 unix_dgram_recv_handler, ctx);
225 if (ctx->sock_read_watch == NULL) {
242 static void unix_dgram_recv_handler(struct poll_watch *w, int fd, short events,
245 struct unix_dgram_ctx *ctx = (struct unix_dgram_ctx *)private_data;
250 #ifdef HAVE_STRUCT_MSGHDR_MSG_CONTROL
251 char buf[CMSG_SPACE(sizeof(int)*INT8_MAX)] = { 0, };
252 struct cmsghdr *cmsg;
253 #endif /* HAVE_STRUCT_MSGHDR_MSG_CONTROL */
255 size_t i, num_fds = 0;
257 iov = (struct iovec) {
258 .iov_base = (void *)ctx->recv_buf,
259 .iov_len = ctx->max_msg,
262 msg = (struct msghdr) {
265 #ifdef HAVE_STRUCT_MSGHDR_MSG_CONTROL
267 .msg_controllen = sizeof(buf),
271 #ifdef MSG_CMSG_CLOEXEC
272 flags |= MSG_CMSG_CLOEXEC;
275 received = recvmsg(fd, &msg, flags);
276 if (received == -1) {
277 if ((errno == EAGAIN) ||
278 (errno == EWOULDBLOCK) ||
279 (errno == EINTR) || (errno == ENOMEM)) {
280 /* Not really an error - just try again. */
283 /* Problem with the socket. Set it unreadable. */
284 ctx->ev_funcs->watch_update(w, 0);
287 if (received > ctx->max_msg) {
288 /* More than we expected, not for us */
292 #ifdef HAVE_STRUCT_MSGHDR_MSG_CONTROL
293 for(cmsg = CMSG_FIRSTHDR(&msg); cmsg != NULL;
294 cmsg = CMSG_NXTHDR(&msg, cmsg))
296 void *data = CMSG_DATA(cmsg);
298 if (cmsg->cmsg_type != SCM_RIGHTS) {
301 if (cmsg->cmsg_level != SOL_SOCKET) {
306 num_fds = (cmsg->cmsg_len - CMSG_LEN(0)) / sizeof (int);
311 for (i = 0; i < num_fds; i++) {
314 err = prepare_socket_cloexec(fds[i]);
320 ctx->recv_callback(ctx, ctx->recv_buf, received,
321 fds, num_fds, ctx->private_data);
325 close_fd_array(fds, num_fds);
327 ctx->recv_callback(ctx, ctx->recv_buf, received,
328 NULL, 0, ctx->private_data);
331 static void unix_dgram_job_finished(struct poll_watch *w, int fd, short events,
334 static int unix_dgram_init_pthreadpool(struct unix_dgram_ctx *ctx)
338 if (ctx->send_pool != NULL) {
342 ret = pthreadpool_init(0, &ctx->send_pool);
347 signalfd = pthreadpool_signal_fd(ctx->send_pool);
349 ctx->pool_read_watch = ctx->ev_funcs->watch_new(
350 ctx->ev_funcs, signalfd, POLLIN,
351 unix_dgram_job_finished, ctx);
352 if (ctx->pool_read_watch == NULL) {
353 pthreadpool_destroy(ctx->send_pool);
354 ctx->send_pool = NULL;
361 static int unix_dgram_send_queue_init(
362 struct unix_dgram_ctx *ctx, const struct sockaddr_un *dst,
363 struct unix_dgram_send_queue **result)
365 struct unix_dgram_send_queue *q;
369 pathlen = strlen(dst->sun_path)+1;
371 q = malloc(offsetof(struct unix_dgram_send_queue, path) + pathlen);
377 memcpy(q->path, dst->sun_path, pathlen);
379 q->sock = socket(AF_UNIX, SOCK_DGRAM, 0);
385 err = prepare_socket_cloexec(q->sock);
391 ret = connect(q->sock,
392 (const struct sockaddr *)(const void *)dst,
394 } while ((ret == -1) && (errno == EINTR));
401 err = unix_dgram_init_pthreadpool(ctx);
406 DLIST_ADD(ctx->send_queues, q);
418 static void unix_dgram_send_queue_free(struct unix_dgram_send_queue *q)
420 struct unix_dgram_ctx *ctx = q->ctx;
422 while (q->msgs != NULL) {
423 struct unix_dgram_msg *msg;
425 DLIST_REMOVE(q->msgs, msg);
426 close_fd_array(msg->fds, msg->num_fds);
430 DLIST_REMOVE(ctx->send_queues, q);
434 static struct unix_dgram_send_queue *find_send_queue(
435 struct unix_dgram_ctx *ctx, const char *dst_sock)
437 struct unix_dgram_send_queue *s;
439 for (s = ctx->send_queues; s != NULL; s = s->next) {
440 if (strcmp(s->path, dst_sock) == 0) {
447 static int queue_msg(struct unix_dgram_send_queue *q,
448 const struct iovec *iov, int iovlen,
449 const int *fds, size_t num_fds)
451 struct unix_dgram_msg *msg;
454 size_t fds_size = sizeof(int) * num_fds;
455 int fds_copy[MIN(num_fds, INT8_MAX)];
456 size_t fds_padding = 0;
461 if (num_fds > INT8_MAX) {
465 for (i = 0; i < num_fds; i++) {
469 for (i = 0; i < num_fds; i++) {
470 fds_copy[i] = dup(fds[i]);
471 if (fds_copy[i] == -1) {
477 buflen = iov_buflen(iov, iovlen);
482 msglen = offsetof(struct unix_dgram_msg, buf);
483 tmp = msglen + buflen;
484 if ((tmp < msglen) || (tmp < buflen)) {
491 const size_t fds_align = sizeof(int) - 1;
493 tmp = msglen + fds_align;
494 if ((tmp < msglen) || (tmp < fds_align)) {
500 fds_padding = tmp - msglen;
503 tmp = msglen + fds_size;
504 if ((tmp < msglen) || (tmp < fds_size)) {
511 msg = malloc(msglen);
516 msg->buflen = buflen;
520 for (i=0; i<iovlen; i++) {
521 memcpy(&msg->buf[buflen], iov[i].iov_base, iov[i].iov_len);
522 buflen += iov[i].iov_len;
525 msg->num_fds = num_fds;
526 if (msg->num_fds > 0) {
527 void *fds_ptr = (void *)&msg->buf[buflen+fds_padding];
528 memcpy(fds_ptr, fds_copy, fds_size);
529 msg->fds = (int *)fds_ptr;
534 DLIST_ADD_END(q->msgs, msg, struct unix_dgram_msg);
540 close_fd_array(fds_copy, num_fds);
544 static void unix_dgram_send_job(void *private_data)
546 struct unix_dgram_msg *dmsg = private_data;
548 .iov_base = (void *)dmsg->buf,
549 .iov_len = dmsg->buflen,
551 struct msghdr msg = {
555 #ifdef HAVE_STRUCT_MSGHDR_MSG_CONTROL
556 struct cmsghdr *cmsg;
557 size_t fds_size = sizeof(int) * dmsg->num_fds;
558 size_t cmsg_len = CMSG_LEN(fds_size);
559 size_t cmsg_space = CMSG_SPACE(fds_size);
560 char cmsg_buf[cmsg_space];
562 if (dmsg->num_fds > 0) {
565 memset(cmsg_buf, 0, cmsg_space);
567 msg.msg_control = cmsg_buf;
568 msg.msg_controllen = cmsg_space;
569 cmsg = CMSG_FIRSTHDR(&msg);
570 cmsg->cmsg_level = SOL_SOCKET;
571 cmsg->cmsg_type = SCM_RIGHTS;
572 cmsg->cmsg_len = cmsg_len;
573 fdptr = CMSG_DATA(cmsg);
574 memcpy(fdptr, dmsg->fds, fds_size);
575 msg.msg_controllen = cmsg->cmsg_len;
577 #endif /* HAVE_STRUCT_MSGHDR_MSG_CONTROL */
580 dmsg->sent = sendmsg(dmsg->sock, &msg, 0);
581 } while ((dmsg->sent == -1) && (errno == EINTR));
583 close_fd_array(dmsg->fds, dmsg->num_fds);
586 static void unix_dgram_job_finished(struct poll_watch *w, int fd, short events,
589 struct unix_dgram_ctx *ctx = private_data;
590 struct unix_dgram_send_queue *q;
591 struct unix_dgram_msg *msg;
594 ret = pthreadpool_finished_jobs(ctx->send_pool, &job, 1);
599 for (q = ctx->send_queues; q != NULL; q = q->next) {
600 if (job == q->sock) {
606 /* Huh? Should not happen */
611 DLIST_REMOVE(q->msgs, msg);
612 close_fd_array(msg->fds, msg->num_fds);
615 if (q->msgs != NULL) {
616 ret = pthreadpool_add_job(ctx->send_pool, q->sock,
617 unix_dgram_send_job, q->msgs);
623 unix_dgram_send_queue_free(q);
626 static int unix_dgram_send(struct unix_dgram_ctx *ctx,
627 const struct sockaddr_un *dst,
628 const struct iovec *iov, int iovlen,
629 const int *fds, size_t num_fds)
631 struct unix_dgram_send_queue *q;
633 #ifdef HAVE_STRUCT_MSGHDR_MSG_CONTROL
634 struct cmsghdr *cmsg;
635 size_t fds_size = sizeof(int) * num_fds;
636 size_t cmsg_len = CMSG_LEN(fds_size);
637 size_t cmsg_space = CMSG_SPACE(fds_size);
638 char cmsg_buf[cmsg_space];
639 #endif /* HAVE_STRUCT_MSGHDR_MSG_CONTROL */
643 if (num_fds > INT8_MAX) {
647 #ifndef HAVE_STRUCT_MSGHDR_MSG_CONTROL
651 #endif /* ! HAVE_STRUCT_MSGHDR_MSG_CONTROL */
653 for (i = 0; i < num_fds; i++) {
655 * Make sure we only allow fd passing
656 * for communication channels,
657 * e.g. sockets, pipes, fifos, ...
659 ret = lseek(fds[i], 0, SEEK_CUR);
660 if (ret == -1 && errno == ESPIPE) {
666 * Reject the message as we may need to call dup(),
667 * if we queue the message.
669 * That might result in unexpected behavior for the caller
670 * for files and broken posix locking.
676 * To preserve message ordering, we have to queue a message when
677 * others are waiting in line already.
679 q = find_send_queue(ctx, dst->sun_path);
681 return queue_msg(q, iov, iovlen, fds, num_fds);
685 * Try a cheap nonblocking send
688 msg = (struct msghdr) {
689 .msg_name = discard_const_p(struct sockaddr_un, dst),
690 .msg_namelen = sizeof(*dst),
691 .msg_iov = discard_const_p(struct iovec, iov),
694 #ifdef HAVE_STRUCT_MSGHDR_MSG_CONTROL
698 memset(cmsg_buf, 0, cmsg_space);
700 msg.msg_control = cmsg_buf;
701 msg.msg_controllen = cmsg_space;
702 cmsg = CMSG_FIRSTHDR(&msg);
703 cmsg->cmsg_level = SOL_SOCKET;
704 cmsg->cmsg_type = SCM_RIGHTS;
705 cmsg->cmsg_len = cmsg_len;
706 fdptr = CMSG_DATA(cmsg);
707 memcpy(fdptr, fds, fds_size);
708 msg.msg_controllen = cmsg->cmsg_len;
710 #endif /* HAVE_STRUCT_MSGHDR_MSG_CONTROL */
712 ret = sendmsg(ctx->sock, &msg, 0);
716 if ((errno != EWOULDBLOCK) && (errno != EAGAIN) && (errno != EINTR)) {
720 ret = unix_dgram_send_queue_init(ctx, dst, &q);
724 ret = queue_msg(q, iov, iovlen, fds, num_fds);
726 unix_dgram_send_queue_free(q);
729 ret = pthreadpool_add_job(ctx->send_pool, q->sock,
730 unix_dgram_send_job, q->msgs);
732 unix_dgram_send_queue_free(q);
738 static int unix_dgram_sock(struct unix_dgram_ctx *ctx)
743 static int unix_dgram_free(struct unix_dgram_ctx *ctx)
745 if (ctx->send_queues != NULL) {
749 if (ctx->send_pool != NULL) {
750 int ret = pthreadpool_destroy(ctx->send_pool);
754 ctx->ev_funcs->watch_free(ctx->pool_read_watch);
757 ctx->ev_funcs->watch_free(ctx->sock_read_watch);
759 if (getpid() == ctx->created_pid) {
760 /* If we created it, unlink. Otherwise someone else might
761 * still have it open */
772 * Every message starts with a uint64_t cookie.
774 * A value of 0 indicates a single-fragment message which is complete in
775 * itself. The data immediately follows the cookie.
777 * Every multi-fragment message has a cookie != 0 and starts with a cookie
778 * followed by a struct unix_msg_header and then the data. The pid and sock
779 * fields are used to assure uniqueness on the receiver side.
782 struct unix_msg_hdr {
789 struct unix_msg *prev, *next;
798 struct unix_msg_ctx {
799 struct unix_dgram_ctx *dgram;
803 void (*recv_callback)(struct unix_msg_ctx *ctx,
804 uint8_t *msg, size_t msg_len,
805 int *fds, size_t num_fds,
809 struct unix_msg *msgs;
812 static void unix_msg_recv(struct unix_dgram_ctx *dgram_ctx,
813 uint8_t *buf, size_t buflen,
814 int *fds, size_t num_fds,
817 int unix_msg_init(const struct sockaddr_un *addr,
818 const struct poll_funcs *ev_funcs,
819 size_t fragment_len, uint64_t cookie,
820 void (*recv_callback)(struct unix_msg_ctx *ctx,
821 uint8_t *msg, size_t msg_len,
822 int *fds, size_t num_fds,
825 struct unix_msg_ctx **result)
827 struct unix_msg_ctx *ctx;
830 ctx = malloc(sizeof(*ctx));
835 *ctx = (struct unix_msg_ctx) {
836 .fragment_len = fragment_len,
838 .recv_callback = recv_callback,
839 .private_data = private_data
842 ret = unix_dgram_init(addr, fragment_len, ev_funcs,
843 unix_msg_recv, ctx, &ctx->dgram);
853 int unix_msg_send(struct unix_msg_ctx *ctx, const struct sockaddr_un *dst,
854 const struct iovec *iov, int iovlen,
855 const int *fds, size_t num_fds)
860 struct iovec iov_copy[iovlen+2];
861 struct unix_msg_hdr hdr;
862 struct iovec src_iov;
868 msglen = iov_buflen(iov, iovlen);
873 #ifndef HAVE_STRUCT_MSGHDR_MSG_CONTROL
877 #endif /* ! HAVE_STRUCT_MSGHDR_MSG_CONTROL */
879 if (num_fds > INT8_MAX) {
883 if (msglen <= (ctx->fragment_len - sizeof(uint64_t))) {
886 iov_copy[0].iov_base = &cookie;
887 iov_copy[0].iov_len = sizeof(cookie);
889 memcpy(&iov_copy[1], iov,
890 sizeof(struct iovec) * iovlen);
893 return unix_dgram_send(ctx->dgram, dst, iov_copy, iovlen+1,
897 hdr = (struct unix_msg_hdr) {
900 .sock = unix_dgram_sock(ctx->dgram)
903 iov_copy[0].iov_base = &ctx->cookie;
904 iov_copy[0].iov_len = sizeof(ctx->cookie);
905 iov_copy[1].iov_base = &hdr;
906 iov_copy[1].iov_len = sizeof(hdr);
912 * The following write loop sends the user message in pieces. We have
913 * filled the first two iovecs above with "cookie" and "hdr". In the
914 * following loops we pull message chunks from the user iov array and
915 * fill iov_copy piece by piece, possibly truncating chunks from the
916 * caller's iov array. Ugly, but hopefully efficient.
919 while (sent < msglen) {
921 size_t iov_index = 2;
923 fragment_len = sizeof(ctx->cookie) + sizeof(hdr);
925 while (fragment_len < ctx->fragment_len) {
928 space = ctx->fragment_len - fragment_len;
929 chunk = MIN(space, src_iov.iov_len);
931 iov_copy[iov_index].iov_base = src_iov.iov_base;
932 iov_copy[iov_index].iov_len = chunk;
935 src_iov.iov_base = (char *)src_iov.iov_base + chunk;
936 src_iov.iov_len -= chunk;
937 fragment_len += chunk;
939 if (src_iov.iov_len == 0) {
948 sent += (fragment_len - sizeof(ctx->cookie) - sizeof(hdr));
951 * only the last fragment should pass the fd array.
952 * That simplifies the receiver a lot.
955 ret = unix_dgram_send(ctx->dgram, dst,
959 ret = unix_dgram_send(ctx->dgram, dst,
969 if (ctx->cookie == 0) {
976 static void unix_msg_recv(struct unix_dgram_ctx *dgram_ctx,
977 uint8_t *buf, size_t buflen,
978 int *fds, size_t num_fds,
981 struct unix_msg_ctx *ctx = (struct unix_msg_ctx *)private_data;
982 struct unix_msg_hdr hdr;
983 struct unix_msg *msg;
987 if (buflen < sizeof(cookie)) {
988 close_fd_array(fds, num_fds);
991 memcpy(&cookie, buf, sizeof(cookie));
993 buf += sizeof(cookie);
994 buflen -= sizeof(cookie);
997 ctx->recv_callback(ctx, buf, buflen, fds, num_fds, ctx->private_data);
1001 if (buflen < sizeof(hdr)) {
1002 close_fd_array(fds, num_fds);
1005 memcpy(&hdr, buf, sizeof(hdr));
1008 buflen -= sizeof(hdr);
1010 for (msg = ctx->msgs; msg != NULL; msg = msg->next) {
1011 if ((msg->sender_pid == hdr.pid) &&
1012 (msg->sender_sock == hdr.sock)) {
1017 if ((msg != NULL) && (msg->cookie != cookie)) {
1018 DLIST_REMOVE(ctx->msgs, msg);
1024 msg = malloc(offsetof(struct unix_msg, buf) + hdr.msglen);
1026 close_fd_array(fds, num_fds);
1029 *msg = (struct unix_msg) {
1030 .msglen = hdr.msglen,
1031 .sender_pid = hdr.pid,
1032 .sender_sock = hdr.sock,
1035 DLIST_ADD(ctx->msgs, msg);
1038 space = msg->msglen - msg->received;
1039 if (buflen > space) {
1040 close_fd_array(fds, num_fds);
1044 memcpy(msg->buf + msg->received, buf, buflen);
1045 msg->received += buflen;
1047 if (msg->received < msg->msglen) {
1048 close_fd_array(fds, num_fds);
1052 DLIST_REMOVE(ctx->msgs, msg);
1053 ctx->recv_callback(ctx, msg->buf, msg->msglen, fds, num_fds, ctx->private_data);
1057 int unix_msg_free(struct unix_msg_ctx *ctx)
1061 ret = unix_dgram_free(ctx->dgram);
1066 while (ctx->msgs != NULL) {
1067 struct unix_msg *msg = ctx->msgs;
1068 DLIST_REMOVE(ctx->msgs, msg);
1076 static ssize_t iov_buflen(const struct iovec *iov, int iovlen)
1081 for (i=0; i<iovlen; i++) {
1082 size_t thislen = iov[i].iov_len;
1083 size_t tmp = buflen + thislen;
1085 if ((tmp < buflen) || (tmp < thislen)) {