blob: cc92753792005a5fb46760fc8c25e8a99ad2ea9b [file] [log] [blame]
/*
* Socket Pair protocol layer (sockpair)
*
* Copyright HAProxy Technologies - William Lallemand <wlallemand@haproxy.com>
*
* This program is free software; you can redistribute it and/or
* modify it under the terms of the GNU General Public License
* as published by the Free Software Foundation; either version
* 2 of the License, or (at your option) any later version.
*
*/
#include <ctype.h>
#include <errno.h>
#include <fcntl.h>
#include <pwd.h>
#include <grp.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <syslog.h>
#include <time.h>
#include <sys/socket.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/un.h>
#include <haproxy/api.h>
#include <haproxy/connection.h>
#include <haproxy/errors.h>
#include <haproxy/fd.h>
#include <haproxy/freq_ctr.h>
#include <haproxy/global.h>
#include <haproxy/list.h>
#include <haproxy/listener.h>
#include <haproxy/log.h>
#include <haproxy/protocol.h>
#include <haproxy/proto_sockpair.h>
#include <haproxy/sock.h>
#include <haproxy/time.h>
#include <haproxy/tools.h>
#include <haproxy/version.h>
static void sockpair_add_listener(struct listener *listener, int port);
static int sockpair_bind_listener(struct listener *listener, char *errmsg, int errlen);
static void sockpair_enable_listener(struct listener *listener);
static void sockpair_disable_listener(struct listener *listener);
static int sockpair_connect_server(struct connection *conn, int flags);
static int sockpair_accepting_conn(const struct receiver *rx);
struct connection *sockpair_accept_conn(struct listener *l, int *status);
struct proto_fam proto_fam_sockpair = {
.name = "sockpair",
.sock_domain = AF_CUST_SOCKPAIR,
.sock_family = AF_UNIX,
.sock_addrlen = sizeof(struct sockaddr_un),
.l3_addrlen = sizeof(((struct sockaddr_un*)0)->sun_path),
.addrcmp = NULL,
.bind = sockpair_bind_receiver,
.get_src = NULL,
.get_dst = NULL,
};
/* Note: must not be declared <const> as its list will be overwritten */
static struct protocol proto_sockpair = {
.name = "sockpair",
.fam = &proto_fam_sockpair,
.ctrl_type = SOCK_STREAM,
.sock_domain = AF_CUST_SOCKPAIR,
.sock_type = SOCK_STREAM,
.sock_prot = 0,
.add = sockpair_add_listener,
.listen = sockpair_bind_listener,
.enable = sockpair_enable_listener,
.disable = sockpair_disable_listener,
.unbind = default_unbind_listener,
.accept_conn = sockpair_accept_conn,
.rx_unbind = sock_unbind,
.rx_enable = sock_enable,
.rx_disable = sock_disable,
.rx_listening = sockpair_accepting_conn,
.default_iocb = &sock_accept_iocb,
.connect = &sockpair_connect_server,
.receivers = LIST_HEAD_INIT(proto_sockpair.receivers),
.nb_receivers = 0,
};
INITCALL1(STG_REGISTER, protocol_register, &proto_sockpair);
/* Add <listener> to the list of sockpair listeners (port is ignored). The
* listener's state is automatically updated from LI_INIT to LI_ASSIGNED.
* The number of listeners for the protocol is updated.
*
* Must be called with proto_lock held.
*
*/
static void sockpair_add_listener(struct listener *listener, int port)
{
if (listener->state != LI_INIT)
return;
listener_set_state(listener, LI_ASSIGNED);
listener->rx.proto = &proto_sockpair;
LIST_ADDQ(&proto_sockpair.receivers, &listener->rx.proto_list);
proto_sockpair.nb_receivers++;
}
/* Enable receipt of incoming connections for listener <l>. The receiver must
* still be valid.
*/
static void sockpair_enable_listener(struct listener *l)
{
fd_want_recv_safe(l->rx.fd);
}
/* Disable receipt of incoming connections for listener <l>. The receiver must
* still be valid.
*/
static void sockpair_disable_listener(struct listener *l)
{
fd_stop_recv(l->rx.fd);
}
/* Binds receiver <rx>, and assigns rx->iocb and rx->owner as the callback
* and context, respectively, with ->bind_thread as the thread mask. Returns an
* error code made of ERR_* bits on failure or ERR_NONE on success. On failure,
* an error message may be passed into <errmsg>. Note that the binding address
* is only an FD to receive the incoming FDs on. Thus by definition there is no
* real "bind" operation, this only completes the receiver. Such FDs are not
* inherited upon reload.
*/
int sockpair_bind_receiver(struct receiver *rx, char **errmsg)
{
int err;
/* ensure we never return garbage */
if (errmsg)
*errmsg = 0;
err = ERR_NONE;
if (rx->flags & RX_F_BOUND)
return ERR_NONE;
if (rx->fd == -1) {
err |= ERR_FATAL | ERR_ALERT;
memprintf(errmsg, "sockpair may be only used with inherited FDs");
goto bind_return;
}
if (rx->fd >= global.maxsock) {
err |= ERR_FATAL | ERR_ABORT | ERR_ALERT;
memprintf(errmsg, "not enough free sockets (raise '-n' parameter)");
goto bind_close_return;
}
if (fcntl(rx->fd, F_SETFL, O_NONBLOCK) == -1) {
err |= ERR_FATAL | ERR_ALERT;
memprintf(errmsg, "cannot make socket non-blocking");
goto bind_close_return;
}
rx->flags |= RX_F_BOUND;
fd_insert(rx->fd, rx->owner, rx->iocb, thread_mask(rx->settings->bind_thread) & all_threads_mask);
return err;
bind_return:
if (errmsg && *errmsg)
memprintf(errmsg, "%s [fd %d]", *errmsg, rx->fd);
return err;
bind_close_return:
close(rx->fd);
goto bind_return;
}
/* This function changes the state from ASSIGNED to LISTEN. The socket is NOT
* enabled for polling. The return value is composed from ERR_NONE,
* ERR_RETRYABLE and ERR_FATAL. It may return a warning or an error message in
* <errmsg> if the message is at most <errlen> bytes long (including '\0').
* Note that <errmsg> may be NULL if <errlen> is also zero.
*/
static int sockpair_bind_listener(struct listener *listener, char *errmsg, int errlen)
{
int err;
char *msg = NULL;
err = ERR_NONE;
/* ensure we never return garbage */
if (errlen)
*errmsg = 0;
if (listener->state != LI_ASSIGNED)
return ERR_NONE; /* already bound */
if (!(listener->rx.flags & RX_F_BOUND)) {
msg = "receiving socket not bound";
goto err_return;
}
listener_set_state(listener, LI_LISTEN);
return err;
err_return:
if (msg && errlen)
snprintf(errmsg, errlen, "%s [fd %d]", msg, listener->rx.fd);
return err;
}
/*
* Send FD over a unix socket
*
* <send_fd> is the FD to send
* <fd> is the fd of the unix socket to use for the transfer
*
* The iobuf variable could be use in the future to enhance the protocol.
*/
int send_fd_uxst(int fd, int send_fd)
{
char iobuf[2];
struct iovec iov;
struct msghdr msghdr;
char cmsgbuf[CMSG_SPACE(sizeof(int))];
char buf[CMSG_SPACE(sizeof(int))];
struct cmsghdr *cmsg = (void *)buf;
int *fdptr;
iov.iov_base = iobuf;
iov.iov_len = sizeof(iobuf);
memset(&msghdr, 0, sizeof(msghdr));
msghdr.msg_iov = &iov;
msghdr.msg_iovlen = 1;
/* Now send the fds */
msghdr.msg_control = cmsgbuf;
msghdr.msg_controllen = CMSG_SPACE(sizeof(int));
cmsg = CMSG_FIRSTHDR(&msghdr);
cmsg->cmsg_len = CMSG_LEN(sizeof(int));
cmsg->cmsg_level = SOL_SOCKET;
cmsg->cmsg_type = SCM_RIGHTS;
fdptr = (int *)CMSG_DATA(cmsg);
memcpy(fdptr, &send_fd, sizeof(send_fd));
if (sendmsg(fd, &msghdr, 0) != sizeof(iobuf)) {
ha_warning("Failed to transfer socket\n");
return 1;
}
return 0;
}
/*
*
* This function works like uxst_connect_server but instead of creating a
* socket and establishing a connection, it creates a pair of connected
* sockets, and send one of them through the destination FD. The destination FD
* is stored in conn->dst->sin_addr.s_addr during configuration parsing.
*
* conn->target may point either to a valid server or to a backend, depending
* on conn->target. Only OBJ_TYPE_PROXY and OBJ_TYPE_SERVER are supported. The
* <data> parameter is a boolean indicating whether there are data waiting for
* being sent or not, in order to adjust data write polling and on some
* platforms. The <delack> argument is ignored.
*
* Note that a pending send_proxy message accounts for data.
*
* It can return one of :
* - SF_ERR_NONE if everything's OK
* - SF_ERR_SRVTO if there are no more servers
* - SF_ERR_SRVCL if the connection was refused by the server
* - SF_ERR_PRXCOND if the connection has been limited by the proxy (maxconn)
* - SF_ERR_RESOURCE if a system resource is lacking (eg: fd limits, ports, ...)
* - SF_ERR_INTERNAL for any other purely internal errors
* Additionally, in the case of SF_ERR_RESOURCE, an emergency log will be emitted.
*
* The connection's fd is inserted only when SF_ERR_NONE is returned, otherwise
* it's invalid and the caller has nothing to do.
*/
static int sockpair_connect_server(struct connection *conn, int flags)
{
int sv[2], fd, dst_fd = -1;
/* the FD is stored in the sockaddr struct */
dst_fd = ((struct sockaddr_in *)conn->dst)->sin_addr.s_addr;
if (obj_type(conn->target) != OBJ_TYPE_PROXY &&
obj_type(conn->target) != OBJ_TYPE_SERVER) {
conn->flags |= CO_FL_ERROR;
return SF_ERR_INTERNAL;
}
if (socketpair(PF_UNIX, SOCK_STREAM, 0, sv) == -1) {
ha_alert("socketpair(): Cannot create socketpair. Giving up.\n");
conn->flags |= CO_FL_ERROR;
return SF_ERR_RESOURCE;
}
fd = conn->handle.fd = sv[1];
if (fd >= global.maxsock) {
/* do not log anything there, it's a normal condition when this option
* is used to serialize connections to a server !
*/
ha_alert("socket(): not enough free sockets. Raise -n argument. Giving up.\n");
close(sv[0]);
close(sv[1]);
conn->err_code = CO_ER_CONF_FDLIM;
conn->flags |= CO_FL_ERROR;
return SF_ERR_PRXCOND; /* it is a configuration limit */
}
if (fcntl(fd, F_SETFL, O_NONBLOCK) == -1) {
qfprintf(stderr,"Cannot set client socket to non blocking mode.\n");
close(sv[0]);
close(sv[1]);
conn->err_code = CO_ER_SOCK_ERR;
conn->flags |= CO_FL_ERROR;
return SF_ERR_INTERNAL;
}
if (master == 1 && (fcntl(fd, F_SETFD, FD_CLOEXEC) == -1)) {
ha_alert("Cannot set CLOEXEC on client socket.\n");
close(sv[0]);
close(sv[1]);
conn->err_code = CO_ER_SOCK_ERR;
conn->flags |= CO_FL_ERROR;
return SF_ERR_INTERNAL;
}
/* if a send_proxy is there, there are data */
if (conn->send_proxy_ofs)
flags |= CONNECT_HAS_DATA;
if (global.tune.server_sndbuf)
setsockopt(fd, SOL_SOCKET, SO_SNDBUF, &global.tune.server_sndbuf, sizeof(global.tune.server_sndbuf));
if (global.tune.server_rcvbuf)
setsockopt(fd, SOL_SOCKET, SO_RCVBUF, &global.tune.server_rcvbuf, sizeof(global.tune.server_rcvbuf));
/* The new socket is sent on the other side, it should be retrieved and
* considered as an 'accept' socket on the server side */
if (send_fd_uxst(dst_fd, sv[0]) == -1) {
close(sv[0]);
close(sv[1]);
conn->err_code = CO_ER_SOCK_ERR;
conn->flags |= CO_FL_ERROR;
return SF_ERR_INTERNAL;
}
close(sv[0]); /* we don't need this side anymore */
conn->flags &= ~CO_FL_WAIT_L4_CONN;
conn->flags |= CO_FL_ADDR_TO_SET;
/* Prepare to send a few handshakes related to the on-wire protocol. */
if (conn->send_proxy_ofs)
conn->flags |= CO_FL_SEND_PROXY;
conn_ctrl_init(conn); /* registers the FD */
fdtab[fd].linger_risk = 0; /* no need to disable lingering */
if (conn_xprt_init(conn) < 0) {
conn_full_close(conn);
conn->flags |= CO_FL_ERROR;
return SF_ERR_RESOURCE;
}
return SF_ERR_NONE; /* connection is OK */
}
/*
* Receives a file descriptor transferred from a unix socket.
*
* Return -1 or a socket fd;
*
* The iobuf variable could be used in the future to enhance the protocol.
*/
int recv_fd_uxst(int sock)
{
struct msghdr msghdr;
struct iovec iov;
char iobuf[2];
char cmsgbuf[CMSG_SPACE(sizeof(int))];
char buf[CMSG_SPACE(sizeof(int))];
struct cmsghdr *cmsg = (void *)buf;
int recv_fd = -1;
int ret = -1;
memset(&msghdr, 0, sizeof(msghdr));
iov.iov_base = iobuf;
iov.iov_len = sizeof(iobuf);
msghdr.msg_iov = &iov;
msghdr.msg_iovlen = 1;
msghdr.msg_control = cmsgbuf;
msghdr.msg_controllen = CMSG_SPACE(sizeof(int));
iov.iov_len = sizeof(iobuf);
iov.iov_base = iobuf;
while (1) {
ret = recvmsg(sock, &msghdr, 0);
if (ret == -1 && errno == EINTR)
continue;
else
break;
}
if (ret == -1)
return ret;
cmsg = CMSG_FIRSTHDR(&msghdr);
if (cmsg && cmsg->cmsg_level == SOL_SOCKET &&
cmsg->cmsg_type == SCM_RIGHTS) {
size_t totlen = cmsg->cmsg_len -
CMSG_LEN(0);
memcpy(&recv_fd, CMSG_DATA(cmsg), totlen);
}
return recv_fd;
}
/* Tests if the receiver supports accepting connections. Returns positive on
* success, 0 if not possible, negative if the socket is non-recoverable. In
* practice zero is never returned since we don't support suspending sockets.
* The real test consists in verifying we have a connected SOCK_STREAM of
* family AF_UNIX.
*/
static int sockpair_accepting_conn(const struct receiver *rx)
{
struct sockaddr sa;
socklen_t len;
int val;
len = sizeof(val);
if (getsockopt(rx->fd, SOL_SOCKET, SO_TYPE, &val, &len) == -1)
return -1;
if (val != SOCK_STREAM)
return -1;
len = sizeof(sa);
if (getsockname(rx->fd, &sa, &len) != 0)
return -1;
if (sa.sa_family != AF_UNIX)
return -1;
len = sizeof(val);
if (getsockopt(rx->fd, SOL_SOCKET, SO_ACCEPTCONN, &val, &len) == -1)
return -1;
/* Note: cannot be a listening socket, must be established */
if (val)
return -1;
return 1;
}
/* Accept an incoming connection from listener <l>, and return it, as well as
* a CO_AC_* status code into <status> if not null. Null is returned on error.
* <l> must be a valid listener with a valid frontend.
*/
struct connection *sockpair_accept_conn(struct listener *l, int *status)
{
struct proxy *p = l->bind_conf->frontend;
struct connection *conn = NULL;
int ret;
int cfd;
if ((cfd = recv_fd_uxst(l->rx.fd)) != -1)
fcntl(cfd, F_SETFL, O_NONBLOCK);
if (likely(cfd != -1)) {
/* Perfect, the connection was accepted */
conn = conn_new(&l->obj_type);
if (!conn)
goto fail_conn;
if (!sockaddr_alloc(&conn->src, NULL, 0))
goto fail_addr;
/* just like with UNIX sockets, only the family is filled */
conn->src->ss_family = AF_UNIX;
conn->handle.fd = cfd;
conn->flags |= CO_FL_ADDR_FROM_SET;
ret = CO_AC_DONE;
goto done;
}
switch (errno) {
case EAGAIN:
ret = CO_AC_DONE; /* nothing more to accept */
if (fdtab[l->rx.fd].ev & (FD_POLL_HUP|FD_POLL_ERR)) {
/* the listening socket might have been disabled in a shared
* process and we're a collateral victim. We'll just pause for
* a while in case it comes back. In the mean time, we need to
* clear this sticky flag.
*/
_HA_ATOMIC_AND(&fdtab[l->rx.fd].ev, ~(FD_POLL_HUP|FD_POLL_ERR));
ret = CO_AC_PAUSE;
}
fd_cant_recv(l->rx.fd);
break;
case EINVAL:
/* might be trying to accept on a shut fd (eg: soft stop) */
ret = CO_AC_PAUSE;
break;
case EINTR:
case ECONNABORTED:
ret = CO_AC_RETRY;
break;
case ENFILE:
if (p)
send_log(p, LOG_EMERG,
"Proxy %s reached system FD limit (maxsock=%d). Please check system tunables.\n",
p->id, global.maxsock);
ret = CO_AC_PAUSE;
break;
case EMFILE:
if (p)
send_log(p, LOG_EMERG,
"Proxy %s reached process FD limit (maxsock=%d). Please check 'ulimit-n' and restart.\n",
p->id, global.maxsock);
ret = CO_AC_PAUSE;
break;
case ENOBUFS:
case ENOMEM:
if (p)
send_log(p, LOG_EMERG,
"Proxy %s reached system memory limit (maxsock=%d). Please check system tunables.\n",
p->id, global.maxsock);
ret = CO_AC_PAUSE;
break;
default:
/* unexpected result, let's give up and let other tasks run */
ret = CO_AC_YIELD;
}
done:
if (status)
*status = ret;
return conn;
fail_addr:
conn_free(conn);
conn = NULL;
fail_conn:
ret = CO_AC_PAUSE;
goto done;
}
/*
* Local variables:
* c-indent-level: 8
* c-basic-offset: 8
* End:
*/