From: Hu Tao <hutao@cn.fujitsu.com> * src/util/threadpool.c, src/util/threadpool.h: Thread pool implementation * src/Makefile.am: Build thread pool --- src/Makefile.am | 1 + src/util/threadpool.c | 178 +++++++++++++++++++++++++++++++++++++++++++++++++ src/util/threadpool.h | 23 ++++++ 3 files changed, 202 insertions(+), 0 deletions(-) create mode 100644 src/util/threadpool.c create mode 100644 src/util/threadpool.h diff --git a/src/Makefile.am b/src/Makefile.am index a9a1986..d71c644 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -73,6 +73,7 @@ UTIL_SOURCES = \ util/threads.c util/threads.h \ util/threads-pthread.h \ util/threads-win32.h \ + util/threadpool.c util/threadpool.h \ util/uuid.c util/uuid.h \ util/util.c util/util.h \ util/xml.c util/xml.h \ diff --git a/src/util/threadpool.c b/src/util/threadpool.c new file mode 100644 index 0000000..cf998bf --- /dev/null +++ b/src/util/threadpool.c @@ -0,0 +1,178 @@ + +#include <config.h> + +#include "threadpool.h" +#include "memory.h" +#include "threads.h" +#include "virterror_internal.h" + +#define VIR_FROM_THIS VIR_FROM_NONE + +typedef struct _virThreadPoolJob virThreadPoolJob; +typedef virThreadPoolJob *virThreadPoolJobPtr; + +struct _virThreadPoolJob { + virThreadPoolJobPtr next; + + void *data; +}; + +struct _virThreadPool { + int quit; + + virThreadPoolJobFunc jobFunc; + void *jobOpaque; + virThreadPoolJobPtr jobList; + + virMutex mutex; + virCond cond; + virCond quit_cond; + + size_t maxWorkers; + size_t freeWorkers; + size_t nWorkers; + virThreadPtr workers; +}; + +static void virThreadPoolWorker(void *opaque) +{ + virThreadPoolPtr pool = opaque; + + virMutexLock(&pool->mutex); + + while (1) { + while (!pool->quit && + !pool->jobList) { + pool->freeWorkers++; + if (virCondWait(&pool->cond, &pool->mutex) < 0) { + pool->freeWorkers--; + break; + } + pool->freeWorkers--; + } + + if (pool->quit) + break; + + virThreadPoolJobPtr job = pool->jobList; + pool->jobList = pool->jobList->next; + job->next = NULL; + + virMutexUnlock(&pool->mutex); + (pool->jobFunc)(job->data, pool->jobOpaque); + VIR_FREE(job); + virMutexLock(&pool->mutex); + } + + pool->nWorkers--; + if (pool->nWorkers == 0) + virCondSignal(&pool->quit_cond); + virMutexUnlock(&pool->mutex); +} + +virThreadPoolPtr virThreadPoolNew(size_t minWorkers, + size_t maxWorkers, + virThreadPoolJobFunc func, + void *opaque) +{ + virThreadPoolPtr pool; + int i; + + if (VIR_ALLOC(pool) < 0) { + virReportOOMError(); + return NULL; + } + + pool->jobFunc = func; + pool->jobOpaque = opaque; + + if (virMutexInit(&pool->mutex) < 0) + goto error; + if (virCondInit(&pool->cond) < 0) + goto error; + if (virCondInit(&pool->quit_cond) < 0) + goto error; + + if (VIR_ALLOC_N(pool->workers, minWorkers) < 0) + goto error; + + pool->maxWorkers = maxWorkers; + for (i = 0; i < minWorkers; i++) { + if (virThreadCreate(&pool->workers[i], + true, + virThreadPoolWorker, + pool) < 0) + goto error; + pool->nWorkers++; + } + + return pool; + +error: + virThreadPoolFree(pool); + return NULL; +} + +void virThreadPoolFree(virThreadPoolPtr pool) +{ + virMutexLock(&pool->mutex); + pool->quit = 1; + if (pool->nWorkers > 0) { + virCondBroadcast(&pool->cond); + if (virCondWait(&pool->quit_cond, &pool->mutex) < 0) + {} + } + VIR_FREE(pool->workers); + virMutexUnlock(&pool->mutex); + VIR_FREE(pool); +} + +int virThreadPoolSendJob(virThreadPoolPtr pool, + void *jobData) +{ + virThreadPoolJobPtr job; + virThreadPoolJobPtr tmp; + + virMutexLock(&pool->mutex); + if (pool->quit) + goto error; + + if (pool->freeWorkers == 0 && + pool->nWorkers < pool->maxWorkers) { + if (VIR_EXPAND_N(pool->workers, pool->nWorkers, 1) < 0) { + virReportOOMError(); + goto error; + } + + if (virThreadCreate(&pool->workers[pool->nWorkers], + true, + virThreadPoolWorker, + pool) < 0) + goto error; + pool->nWorkers++; + } + + if (VIR_ALLOC(job) < 0) { + virReportOOMError(); + goto error; + } + + job->data = jobData; + + tmp = pool->jobList; + while (tmp && tmp->next) + tmp = tmp->next; + if (tmp) + tmp->next = job; + else + pool->jobList = job; + + virCondSignal(&pool->cond); + virMutexUnlock(&pool->mutex); + + return 0; + +error: + virMutexUnlock(&pool->mutex); + return -1; +} diff --git a/src/util/threadpool.h b/src/util/threadpool.h new file mode 100644 index 0000000..093786f --- /dev/null +++ b/src/util/threadpool.h @@ -0,0 +1,23 @@ +#ifndef __THREADPOOL_H__ +#define __THREADPOOL_H__ + +#include <pthread.h> + +typedef struct _virThreadPool virThreadPool; +typedef virThreadPool *virThreadPoolPtr; + +typedef void (*virThreadPoolJobFunc)(void *jobdata, void *opaque); + +virThreadPoolPtr virThreadPoolNew(size_t minWorkers, + size_t maxWorkers, + virThreadPoolJobFunc func, + void *opaque); + +void virThreadPoolShutdown(virThreadPoolPtr pool); + +void virThreadPoolFree(virThreadPoolPtr pool); + +int virThreadPoolSendJob(virThreadPoolPtr pool, + void *jobdata); + +#endif -- 1.7.2.3

The logging setup requires const char * strings, but the virLogSetFromEnv() strdup's the env variables, thus causing a memory leak * src/util/logging.c: Avoid strdup'ing env variables --- src/util/logging.c | 4 ++-- 1 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/util/logging.c b/src/util/logging.c index d65dec0..83cc358 100644 --- a/src/util/logging.c +++ b/src/util/logging.c @@ -980,8 +980,8 @@ void virLogSetFromEnv(void) { virLogParseDefaultPriority(debugEnv); debugEnv = getenv("LIBVIRT_LOG_FILTERS"); if (debugEnv && *debugEnv) - virLogParseFilters(strdup(debugEnv)); + virLogParseFilters(debugEnv); debugEnv = getenv("LIBVIRT_LOG_OUTPUTS"); if (debugEnv && *debugEnv) - virLogParseOutputs(strdup(debugEnv)); + virLogParseOutputs(debugEnv); } -- 1.7.2.3

Introduces a set of generic objects which are to be used in building RPC servers/clients based on XDR. - virNetMessageHeader - standardize the XDR format for any RPC program. Copied from remote protocol for back compat - virNetMessage - Provides a buffer for (de-)serializing messages, and a copy of the decoded virNetMessageHeader. Provides APIs for encoding/decoding message headers and payloads, thus isolating all the XDR api calls in one file. Callers no longer need to use XDR themselves. - virNetSocket - a wrapper around a socket file descriptor, to simplify creation of new sockets, both for clients and services. Encapsulates all the hairy getaddrinfo code and sockaddr manipulation. Will eventually include transparent support for TLS and SASL encoding of data - virNetTLSContext - encapsulates the credentials required to setup TLS sessions. eg the set of x509 certificates and keys, optional DH parameters and x509 DName whitelist Provides APIs for easily validating certificates from a TLS session - virNetTLSSession - encapsulates the TLS session handling, so that callers no longer have a direct dependancy on gnutls. This will facilitate adding alternate TLS impls. Makes the read/write TLS functions work with same semantics as the native socket read/write functions. ie they set errno, instead of a gnutls specific error code. This code is taken from either the daemon/libvirtd.c, daemon/dispatch.c or src/remote/remote_driver.c files, which all duplicated alot of functionality. --- src/Makefile.am | 42 +++- src/rpc/virnetmessage.c | 215 +++++++++++++ src/rpc/virnetmessage.h | 31 ++ src/rpc/virnetprotocol.c | 108 +++++++ src/rpc/virnetprotocol.h | 81 +++++ src/rpc/virnetprotocol.x | 162 ++++++++++ src/rpc/virnetsocket.c | 715 ++++++++++++++++++++++++++++++++++++++++++++ src/rpc/virnetsocket.h | 97 ++++++ src/rpc/virnettlscontext.c | 611 +++++++++++++++++++++++++++++++++++++ src/rpc/virnettlscontext.h | 63 ++++ 10 files changed, 2124 insertions(+), 1 deletions(-) create mode 100644 src/rpc/virnetmessage.c create mode 100644 src/rpc/virnetmessage.h create mode 100644 src/rpc/virnetprotocol.c create mode 100644 src/rpc/virnetprotocol.h create mode 100644 src/rpc/virnetprotocol.x create mode 100644 src/rpc/virnetsocket.c create mode 100644 src/rpc/virnetsocket.h create mode 100644 src/rpc/virnettlscontext.c create mode 100644 src/rpc/virnettlscontext.h diff --git a/src/Makefile.am b/src/Makefile.am index d71c644..613ff0a 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -530,6 +530,24 @@ else mv -f rp_qemu.c-t $(srcdir)/remote/qemu_protocol.c endif +rpcgen-net: + rm -f rp_net.c-t rp_net.h-t rp_net.c-t1 rp_net.c-t2 rp_net.h-t1 + $(RPCGEN) -h -o rp_net.h-t $(srcdir)/rpc/virnetprotocol.x + $(RPCGEN) -c -o rp_net.c-t $(srcdir)/rpc/virnetprotocol.x +if HAVE_GLIBC_RPCGEN + perl -w $(srcdir)/remote/rpcgen_fix.pl rp_net.h-t > rp_net.h-t1 + perl -w $(srcdir)/remote/rpcgen_fix.pl rp_net.c-t > rp_net.c-t1 + (echo '#include <config.h>'; cat rp_net.c-t1) > rp_net.c-t2 + chmod 0444 rp_net.c-t2 rp_net.h-t1 + mv -f rp_net.h-t1 $(srcdir)/rpc/virnetprotocol.h + mv -f rp_net.c-t2 $(srcdir)/rpc/virnetprotocol.c + rm -f rp_net.c-t rp_net.h-t rp_net.c-t1 +else + chmod 0444 rp_net.c-t rp_net.h-t + mv -f rp_net.h-t $(srcdir)/rpc/virnetprotocol.h + mv -f rp_net.c-t $(srcdir)/rpc/virnetprotocol.c +endif + # # Maintainer-only target for re-generating the derived .c/.h source # files, which are actually derived from the .x file. @@ -540,7 +558,7 @@ endif # Support for non-GLIB rpcgen is here as a convenience for # non-Linux people needing to test changes during dev. # -rpcgen: rpcgen-normal rpcgen-qemu +rpcgen: rpcgen-normal rpcgen-qemu rpcgen-net endif @@ -1098,6 +1116,28 @@ libvirt_qemu_la_CFLAGS = $(AM_CFLAGS) libvirt_qemu_la_LIBADD = libvirt.la $(CYGWIN_EXTRA_LIBADD) EXTRA_DIST += $(LIBVIRT_QEMU_SYMBOL_FILE) + +noinst_LTLIBRARIES += libvirt-net-rpc.la + +libvirt_net_rpc_la_SOURCES = \ + ../daemon/event.c \ + rpc/virnetprotocol.h rpc/virnetprotocol.c \ + rpc/virnetmessage.h rpc/virnetmessage.c \ + rpc/virnettlscontext.h rpc/virnettlscontext.c \ + rpc/virnetsocket.h rpc/virnetsocket.c +libvirt_net_rpc_la_CFLAGS = \ + $(GNUTLS_CFLAGS) \ + $(SASL_CFLAGS) \ + $(AM_CFLAGS) +libvirt_net_rpc_la_LDFLAGS = \ + $(GNUTLS_LIBS) \ + $(SASL_LIBS) \ + $(AM_LDFLAGS) \ + $(CYGWIN_EXTRA_LDFLAGS) \ + $(MINGW_EXTRA_LDFLAGS)l +libvirt_net_rpc_la_LIBADD = \ + $(CYGWIN_EXTRA_LIBADD) + libexec_PROGRAMS = if WITH_STORAGE_DISK diff --git a/src/rpc/virnetmessage.c b/src/rpc/virnetmessage.c new file mode 100644 index 0000000..9bd7557 --- /dev/null +++ b/src/rpc/virnetmessage.c @@ -0,0 +1,215 @@ +#include <config.h> + +#include "virnetmessage.h" + +#include "virterror_internal.h" +#include "logging.h" + +#define virNetError(code, ...) \ + virReportErrorHelper(NULL, VIR_FROM_RPC, code, __FILE__, \ + __FUNCTION__, __LINE__, __VA_ARGS__) + + +int virNetMessageDecodeLength(virNetMessagePtr msg) +{ + XDR xdr; + unsigned int len; + int ret = -1; + + xdrmem_create(&xdr, msg->buffer, + msg->bufferLength, XDR_DECODE); + if (!xdr_u_int(&xdr, &len)) { + virNetError(VIR_ERR_RPC, "%s", _("Unable to decode message length")); + goto cleanup; + } + msg->bufferOffset = xdr_getpos(&xdr); + + if (len < VIR_NET_MESSAGE_LEN_MAX) { + virNetError(VIR_ERR_RPC, "%s", + _("packet received from server too small")); + goto cleanup; + } + + /* Length includes length word - adjust to real length to read. */ + len -= VIR_NET_MESSAGE_LEN_MAX; + + if (len > VIR_NET_MESSAGE_MAX) { + virNetError(VIR_ERR_RPC, "%s", + _("packet received from server too large")); + goto cleanup; + } + + /* Extend our declared buffer length and carry + on reading the header + payload */ + msg->bufferLength += len; + + VIR_DEBUG("Got length, now need %d total (%d more)", msg->bufferLength, len); + + ret = 0; + +cleanup: + xdr_destroy(&xdr); + return ret; +} + + +/* + * @msg: the complete incoming message, whose header to decode + * + * Decodes the header part of the message, but does not + * validate the decoded fields in the header. It expects + * bufferLength to refer to length of the data packet. Upon + * return bufferOffset will refer to the amount of the packet + * consumed by decoding of the header. + * + * returns 0 if successfully decoded, -1 upon fatal error + */ +int virNetMessageDecodeHeader(virNetMessagePtr msg) +{ + XDR xdr; + int ret = -1; + + msg->bufferOffset = VIR_NET_MESSAGE_LEN_MAX; + + /* Parse the header. */ + xdrmem_create(&xdr, + msg->buffer + msg->bufferOffset, + msg->bufferLength - msg->bufferOffset, + XDR_DECODE); + + if (!xdr_virNetMessageHeader(&xdr, &msg->header)) { + virNetError(VIR_ERR_RPC, "%s", _("Unable to decode message header")); + goto cleanup; + } + + msg->bufferOffset += xdr_getpos(&xdr); + + ret = 0; + +cleanup: + xdr_destroy(&xdr); + return ret; +} + + +/* + * @msg: the outgoing message, whose header to encode + * + * Encodes the length word and header of the message, setting the + * message offset ready to encode the payload. Leaves space + * for the length field later. Upon return bufferLength will + * refer to the total available space for message, while + * bufferOffset will refer to current space used by header + * + * returns 0 if successfully encoded, -1 upon fatal error + */ +int virNetMessageEncodeHeader(virNetMessagePtr msg) +{ + XDR xdr; + int ret = -1; + unsigned int len = 0; + + msg->bufferLength = sizeof(msg->buffer); + msg->bufferOffset = 0; + + /* Format the header. */ + xdrmem_create(&xdr, + msg->buffer, + msg->bufferLength, + XDR_ENCODE); + + /* The real value is filled in shortly */ + if (!xdr_u_int(&xdr, &len)) { + virNetError(VIR_ERR_RPC, "%s", _("Unable to encode message length")); + goto cleanup; + } + + if (!xdr_virNetMessageHeader(&xdr, &msg->header)) { + virNetError(VIR_ERR_RPC, "%s", _("Unable to encode message header")); + goto cleanup; + } + + len = xdr_getpos(&xdr); + xdr_setpos(&xdr, 0); + + /* Fill in current length - may be re-written later + * if a payload is added + */ + if (!xdr_u_int(&xdr, &len)) { + virNetError(VIR_ERR_RPC, "%s", _("Unable to re-encode message length")); + goto cleanup; + } + + msg->bufferOffset += len; + + ret = 0; + +cleanup: + xdr_destroy(&xdr); + return ret; +} + + +int virNetMessageEncodePayload(virNetMessagePtr msg, + xdrproc_t filter, + void *data) +{ + XDR xdr; + + /* Serialise header followed by args. */ + xdrmem_create(&xdr, msg->buffer + msg->bufferOffset, + msg->bufferLength - msg->bufferOffset, XDR_ENCODE); + + if (!(*filter)(&xdr, data)) { + virNetError(VIR_ERR_RPC, "%s", _("Unable to encode message payload")); + goto error; + } + + /* Get the length stored in buffer. */ + msg->bufferOffset += xdr_getpos(&xdr); + xdr_destroy(&xdr); + + /* Re-encode the length word. */ + VIR_DEBUG("Encode length as %d", msg->bufferLength); + xdrmem_create(&xdr, msg->buffer, VIR_NET_MESSAGE_HEADER_XDR_LEN, XDR_ENCODE); + if (!xdr_u_int(&xdr, &msg->bufferOffset)) { + virNetError(VIR_ERR_RPC, "%s", _("Unable to encode message length")); + goto error; + } + xdr_destroy(&xdr); + + msg->bufferLength = msg->bufferOffset; + msg->bufferOffset = 0; + return 0; + +error: + xdr_destroy(&xdr); + return -1; +} + + +int virNetMessageDecodePayload(virNetMessagePtr msg, + xdrproc_t filter, + void *data) +{ + XDR xdr; + + /* Serialise header followed by args. */ + xdrmem_create(&xdr, msg->buffer + msg->bufferOffset, + msg->bufferLength - msg->bufferOffset, XDR_DECODE); + + if (!(*filter)(&xdr, data)) { + virNetError(VIR_ERR_RPC, "%s", _("Unable to decode message payload")); + goto error; + } + + /* Get the length stored in buffer. */ + msg->bufferLength += xdr_getpos(&xdr); + xdr_destroy(&xdr); + return 0; + +error: + xdr_destroy(&xdr); + return -1; +} + diff --git a/src/rpc/virnetmessage.h b/src/rpc/virnetmessage.h new file mode 100644 index 0000000..73d417b --- /dev/null +++ b/src/rpc/virnetmessage.h @@ -0,0 +1,31 @@ +#ifndef __VIR_NET_MESSAGE_H__ +#define __VIR_NET_MESSAGE_H__ + +#include "virnetprotocol.h" + +typedef struct virNetMessageHeader *virNetMessageHeaderPtr; + +typedef struct _virNetMessage virNetMessage; +typedef virNetMessage *virNetMessagePtr; + +struct _virNetMessage { + char buffer[VIR_NET_MESSAGE_MAX + VIR_NET_MESSAGE_LEN_MAX]; + unsigned int bufferLength; + unsigned int bufferOffset; + + virNetMessageHeader header; +}; + +int virNetMessageEncodeHeader(virNetMessagePtr msg); +int virNetMessageDecodeLength(virNetMessagePtr msg); +int virNetMessageDecodeHeader(virNetMessagePtr msg); + +int virNetMessageEncodePayload(virNetMessagePtr msg, + xdrproc_t filter, + void *data); +int virNetMessageDecodePayload(virNetMessagePtr msg, + xdrproc_t filter, + void *data); + +#endif /* __VIR_NET_MESSAGE_H__ */ + diff --git a/src/rpc/virnetprotocol.c b/src/rpc/virnetprotocol.c new file mode 100644 index 0000000..0a803ae --- /dev/null +++ b/src/rpc/virnetprotocol.c @@ -0,0 +1,108 @@ +#include <config.h> +/* + * Please do not edit this file. + * It was generated using rpcgen. + */ + +#include "./rpc/virnetprotocol.h" +#include "internal.h" +#ifdef HAVE_XDR_U_INT64_T +# define xdr_uint64_t xdr_u_int64_t +#endif +#ifndef IXDR_PUT_INT32 +# define IXDR_PUT_INT32 IXDR_PUT_LONG +#endif +#ifndef IXDR_GET_INT32 +# define IXDR_GET_INT32 IXDR_GET_LONG +#endif +#ifndef IXDR_PUT_U_INT32 +# define IXDR_PUT_U_INT32 IXDR_PUT_U_LONG +#endif +#ifndef IXDR_GET_U_INT32 +# define IXDR_GET_U_INT32 IXDR_GET_U_LONG +#endif + +bool_t +xdr_virNetMessageType (XDR *xdrs, virNetMessageType *objp) +{ + + if (!xdr_enum (xdrs, (enum_t *) objp)) + return FALSE; + return TRUE; +} + +bool_t +xdr_virNetMessageStatus (XDR *xdrs, virNetMessageStatus *objp) +{ + + if (!xdr_enum (xdrs, (enum_t *) objp)) + return FALSE; + return TRUE; +} + +bool_t +xdr_virNetMessageHeader (XDR *xdrs, virNetMessageHeader *objp) +{ + register int32_t *buf; + + + if (xdrs->x_op == XDR_ENCODE) { + buf = (int32_t*)XDR_INLINE (xdrs, 3 * BYTES_PER_XDR_UNIT); + if (buf == NULL) { + if (!xdr_u_int (xdrs, &objp->prog)) + return FALSE; + if (!xdr_u_int (xdrs, &objp->vers)) + return FALSE; + if (!xdr_int (xdrs, &objp->proc)) + return FALSE; + + } else { + (void)IXDR_PUT_U_INT32(buf, objp->prog); + (void)IXDR_PUT_U_INT32(buf, objp->vers); + (void)IXDR_PUT_INT32(buf, objp->proc); + } + if (!xdr_virNetMessageType (xdrs, &objp->type)) + return FALSE; + if (!xdr_u_int (xdrs, &objp->serial)) + return FALSE; + if (!xdr_virNetMessageStatus (xdrs, &objp->status)) + return FALSE; + return TRUE; + } else if (xdrs->x_op == XDR_DECODE) { + buf = (int32_t*)XDR_INLINE (xdrs, 3 * BYTES_PER_XDR_UNIT); + if (buf == NULL) { + if (!xdr_u_int (xdrs, &objp->prog)) + return FALSE; + if (!xdr_u_int (xdrs, &objp->vers)) + return FALSE; + if (!xdr_int (xdrs, &objp->proc)) + return FALSE; + + } else { + objp->prog = IXDR_GET_U_LONG(buf); + objp->vers = IXDR_GET_U_LONG(buf); + objp->proc = IXDR_GET_INT32(buf); + } + if (!xdr_virNetMessageType (xdrs, &objp->type)) + return FALSE; + if (!xdr_u_int (xdrs, &objp->serial)) + return FALSE; + if (!xdr_virNetMessageStatus (xdrs, &objp->status)) + return FALSE; + return TRUE; + } + + if (!xdr_u_int (xdrs, &objp->prog)) + return FALSE; + if (!xdr_u_int (xdrs, &objp->vers)) + return FALSE; + if (!xdr_int (xdrs, &objp->proc)) + return FALSE; + if (!xdr_virNetMessageType (xdrs, &objp->type)) + return FALSE; + if (!xdr_u_int (xdrs, &objp->serial)) + return FALSE; + if (!xdr_virNetMessageStatus (xdrs, &objp->status)) + return FALSE; + return TRUE; +} diff --git a/src/rpc/virnetprotocol.h b/src/rpc/virnetprotocol.h new file mode 100644 index 0000000..4ef8e20 --- /dev/null +++ b/src/rpc/virnetprotocol.h @@ -0,0 +1,81 @@ +/* + * Please do not edit this file. + * It was generated using rpcgen. + */ + +#ifndef _RP_NET_H_RPCGEN +#define _RP_NET_H_RPCGEN + +#include <rpc/rpc.h> + + +#ifdef __cplusplus +extern "C" { +#endif + +#include "internal.h" +#ifdef HAVE_XDR_U_INT64_T +# define xdr_uint64_t xdr_u_int64_t +#endif +#ifndef IXDR_PUT_INT32 +# define IXDR_PUT_INT32 IXDR_PUT_LONG +#endif +#ifndef IXDR_GET_INT32 +# define IXDR_GET_INT32 IXDR_GET_LONG +#endif +#ifndef IXDR_PUT_U_INT32 +# define IXDR_PUT_U_INT32 IXDR_PUT_U_LONG +#endif +#ifndef IXDR_GET_U_INT32 +# define IXDR_GET_U_INT32 IXDR_GET_U_LONG +#endif +#define VIR_NET_MESSAGE_MAX 262144 +#define VIR_NET_MESSAGE_HEADER_MAX 24 +#define VIR_NET_MESSAGE_PAYLOAD_MAX 262120 +#define VIR_NET_MESSAGE_LEN_MAX 4 + +enum virNetMessageType { + VIR_NET_CALL = 0, + VIR_NET_REPLY = 1, + VIR_NET_MESSAGE = 2, + VIR_NET_STREAM = 3, +}; +typedef enum virNetMessageType virNetMessageType; + +enum virNetMessageStatus { + VIR_NET_OK = 0, + VIR_NET_ERROR = 1, + VIR_NET_CONTINUE = 2, +}; +typedef enum virNetMessageStatus virNetMessageStatus; +#define VIR_NET_MESSAGE_HEADER_XDR_LEN 4 + +struct virNetMessageHeader { + u_int prog; + u_int vers; + int proc; + virNetMessageType type; + u_int serial; + virNetMessageStatus status; +}; +typedef struct virNetMessageHeader virNetMessageHeader; + +/* the xdr functions */ + +#if defined(__STDC__) || defined(__cplusplus) +extern bool_t xdr_virNetMessageType (XDR *, virNetMessageType*); +extern bool_t xdr_virNetMessageStatus (XDR *, virNetMessageStatus*); +extern bool_t xdr_virNetMessageHeader (XDR *, virNetMessageHeader*); + +#else /* K&R C */ +extern bool_t xdr_virNetMessageType (); +extern bool_t xdr_virNetMessageStatus (); +extern bool_t xdr_virNetMessageHeader (); + +#endif /* K&R C */ + +#ifdef __cplusplus +} +#endif + +#endif /* !_RP_NET_H_RPCGEN */ diff --git a/src/rpc/virnetprotocol.x b/src/rpc/virnetprotocol.x new file mode 100644 index 0000000..1d74153 --- /dev/null +++ b/src/rpc/virnetprotocol.x @@ -0,0 +1,162 @@ +/* -*- c -*- + * virnetprotocol.x: basic protocol for all RPC services. + * + * Copyright (C) 2006-2010 Red Hat, Inc. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA + * + * Author: Richard Jones <rjones@redhat.com> + */ + +%#include "internal.h" + +/* cygwin's xdr implementation defines xdr_u_int64_t instead of xdr_uint64_t + * and lacks IXDR_PUT_INT32 and IXDR_GET_INT32 + */ +%#ifdef HAVE_XDR_U_INT64_T +%# define xdr_uint64_t xdr_u_int64_t +%#endif +%#ifndef IXDR_PUT_INT32 +%# define IXDR_PUT_INT32 IXDR_PUT_LONG +%#endif +%#ifndef IXDR_GET_INT32 +%# define IXDR_GET_INT32 IXDR_GET_LONG +%#endif +%#ifndef IXDR_PUT_U_INT32 +%# define IXDR_PUT_U_INT32 IXDR_PUT_U_LONG +%#endif +%#ifndef IXDR_GET_U_INT32 +%# define IXDR_GET_U_INT32 IXDR_GET_U_LONG +%#endif + +/*----- Data types. -----*/ + +/* Maximum total message size (serialised). */ +const VIR_NET_MESSAGE_MAX = 262144; + +/* Size of struct virNetMessageHeader (serialized)*/ +const VIR_NET_MESSAGE_HEADER_MAX = 24; + +/* Size of message payload */ +const VIR_NET_MESSAGE_PAYLOAD_MAX = 262120; + +/* Size of message length field. Not counted in VIR_NET_MESSAGE_MAX */ +const VIR_NET_MESSAGE_LEN_MAX = 4; + +/* + * RPC wire format + * + * Each message consists of: + * + * Name | Type | Description + * -----------+-----------------------+------------------ + * Length | int | Total number of bytes in message _including_ length. + * Header | virNetMessageHeader | Control information about procedure call + * Payload | - | Variable payload data per procedure + * + * In header, the 'serial' field varies according to: + * + * - type == VIR_NET_CALL + * * serial is set by client, incrementing by 1 each time + * + * - type == VIR_NET_REPLY + * * serial matches that from the corresponding VIR_NET_CALL + * + * - type == VIR_NET_MESSAGE + * * serial is always zero + * + * - type == VIR_NET_STREAM + * * serial matches that from the corresponding VIR_NET_CALL + * + * and the 'status' field varies according to: + * + * - type == VIR_NET_CALL + * * VIR_NET_OK always + * + * - type == VIR_NET_REPLY + * * VIR_NET_OK if RPC finished successfully + * * VIR_NET_ERROR if something failed + * + * - type == VIR_NET_MESSAGE + * * VIR_NET_OK always + * + * - type == VIR_NET_STREAM + * * VIR_NET_CONTINUE if more data is following + * * VIR_NET_OK if stream is complete + * * VIR_NET_ERROR if stream had an error + * + * Payload varies according to type and status: + * + * - type == VIR_NET_CALL + * XXX_args for procedure + * + * - type == VIR_NET_REPLY + * * status == VIR_NET_OK + * XXX_ret for procedure + * * status == VIR_NET_ERROR + * remote_error Error information + * + * - type == VIR_NET_MESSAGE + * * status == VIR_NET_OK + * XXX_args for procedure + * * status == VIR_NET_ERROR + * remote_error Error information + * + * - type == VIR_NET_STREAM + * * status == VIR_NET_CONTINUE + * byte[] raw stream data + * * status == VIR_NET_ERROR + * remote_error error information + * * status == VIR_NET_OK + * <empty> + */ +enum virNetMessageType { + /* client -> server. args from a method call */ + VIR_NET_CALL = 0, + /* server -> client. reply/error from a method call */ + VIR_NET_REPLY = 1, + /* either direction. async notification */ + VIR_NET_MESSAGE = 2, + /* either direction. stream data packet */ + VIR_NET_STREAM = 3 +}; + +enum virNetMessageStatus { + /* Status is always VIR_NET_OK for calls. + * For replies, indicates no error. + */ + VIR_NET_OK = 0, + + /* For replies, indicates that an error happened, and a struct + * remote_error follows. + */ + VIR_NET_ERROR = 1, + + /* For streams, indicates that more data is still expected + */ + VIR_NET_CONTINUE = 2 +}; + +/* 4 byte length word per header */ +const VIR_NET_MESSAGE_HEADER_XDR_LEN = 4; + +struct virNetMessageHeader { + unsigned prog; /* Unique ID for the program */ + unsigned vers; /* Program version number */ + int proc; /* Unique ID for the procedure within the program */ + virNetMessageType type; /* Type of message */ + unsigned serial; /* Serial number of message. */ + virNetMessageStatus status; +}; diff --git a/src/rpc/virnetsocket.c b/src/rpc/virnetsocket.c new file mode 100644 index 0000000..de66103 --- /dev/null +++ b/src/rpc/virnetsocket.c @@ -0,0 +1,715 @@ +/* + * virnetsocket.h: generic network socket handling + * + * Copyright (C) 2006-2010 Red Hat, Inc. + * Copyright (C) 2006 Daniel P. Berrange + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA + * + * Author: Daniel P. Berrange <berrange@redhat.com> + */ + +#include <config.h> + +#include <sys/stat.h> +#include <unistd.h> +#include <netinet/tcp.h> +#include <sys/wait.h> + +#include "virnetsocket.h" +#include "util.h" +#include "memory.h" +#include "virterror_internal.h" +#include "logging.h" +#include "files.h" +#include "event.h" + +#define VIR_FROM_THIS VIR_FROM_RPC + +#define virNetError(code, ...) \ + virReportErrorHelper(NULL, VIR_FROM_RPC, code, __FILE__, \ + __FUNCTION__, __LINE__, __VA_ARGS__) + + +struct _virNetSocket { + int fd; + int watch; + pid_t pid; + int errfd; + virNetSocketIOFunc func; + void *opaque; + virSocketAddr localAddr; + virSocketAddr remoteAddr; + char *localAddrStr; + char *remoteAddrStr; +}; + + +static int virNetSocketForkDaemon(const char *binary) +{ + const char *const daemonargs[] = { binary, "--timeout=30", NULL }; + pid_t pid; + + if (virExecDaemonize(daemonargs, NULL, NULL, + &pid, -1, NULL, NULL, + VIR_EXEC_CLEAR_CAPS, + NULL, NULL, NULL) < 0) + return -1; + + return 0; +} + + +static virNetSocketPtr virNetSocketNew(virSocketAddrPtr localAddr, + virSocketAddrPtr remoteAddr, + int fd, int errfd, pid_t pid) +{ + virNetSocketPtr sock; + int no_slow_start = 1; + + if (virSetCloseExec(fd) < 0 || + virSetNonBlock(fd) < 0) + return NULL; + + if (VIR_ALLOC(sock) < 0) { + virReportOOMError(); + return NULL; + } + + sock->localAddr = *localAddr; + if (remoteAddr) + sock->remoteAddr = *remoteAddr; + sock->fd = fd; + sock->errfd = errfd; + sock->pid = pid; + + /* Disable nagle */ + if (sock->localAddr.data.sa.sa_family != AF_UNIX) + setsockopt (fd, IPPROTO_TCP, TCP_NODELAY, + (void *)&no_slow_start, + sizeof(no_slow_start)); + + + if (!(sock->localAddrStr = virSocketFormatAddrFull(localAddr, true, ";"))) { + VIR_FREE(sock); + return NULL; + } + + if (remoteAddr && + !(sock->remoteAddrStr = virSocketFormatAddrFull(remoteAddr, true, ";"))) { + VIR_FREE(sock); + return NULL; + } + + VIR_DEBUG("sock=%p", sock); + + return sock; +} + + +int virNetSocketNewListenTCP(const char *nodename, + const char *service, + virNetSocketPtr **retsocks, + size_t *nretsocks) +{ + virNetSocketPtr *socks = NULL; + size_t nsocks = 0; + struct addrinfo *ai; + struct addrinfo hints; + int fd = -1; + + *retsocks = NULL; + *nretsocks = 0; + + memset (&hints, 0, sizeof hints); + hints.ai_flags = AI_PASSIVE | AI_ADDRCONFIG; + hints.ai_socktype = SOCK_STREAM; + + int e = getaddrinfo (nodename, service, &hints, &ai); + if (e != 0) { + virNetError(VIR_ERR_SYSTEM_ERROR, + _("Unable to resolve address '%s' service '%s': %s"), + nodename, service, gai_strerror (e)); + goto error; + } + + struct addrinfo *runp = ai; + while (runp) { + virSocketAddr addr; + + if ((fd = socket(runp->ai_family, runp->ai_socktype, + runp->ai_protocol)) < 0) { + virReportSystemError(errno, "%s", _("Unable to create socket")); + goto error; + } + + int opt = 1; + setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof opt); + +#ifdef IPV6_V6ONLY + if (runp->ai_family == PF_INET6) { + int on = 1; + /* + * Normally on Linux an INET6 socket will bind to the INET4 + * address too. If getaddrinfo returns results with INET4 + * first though, this will result in INET6 binding failing. + * We can trivially cope with multiple server sockets, so + * we force it to only listen on IPv6 + */ + setsockopt(fd, IPPROTO_IPV6,IPV6_V6ONLY, + (void*)&on, sizeof on); + } +#endif + + if (bind(fd, runp->ai_addr, runp->ai_addrlen) < 0) { + if (errno != EADDRINUSE) { + virReportSystemError(errno, "%s", _("Unable to bind to port")); + goto error; + } + VIR_FORCE_CLOSE(fd); + continue; + } + + if (getsockname(fd, &addr.data.sa, &addr.len) < 0) { + virReportSystemError(errno, "%s", _("Unable to get local socket name")); + goto error; + } + + + if (VIR_EXPAND_N(socks, nsocks, 1) < 0) { + virReportOOMError(); + goto error; + } + + if (!(socks[nsocks-1] = virNetSocketNew(&addr, NULL, fd, -1, 0))) + goto error; + runp = runp->ai_next; + } + + freeaddrinfo (ai); + + *retsocks = socks; + *nretsocks = nsocks; + return 0; + +error: + VIR_FORCE_CLOSE(fd); + return -1; +} + +int virNetSocketNewListenUNIX(const char *path, + mode_t mask, + gid_t grp, + virNetSocketPtr *retsock) +{ + virSocketAddr addr; + mode_t oldmask; + gid_t oldgrp; + int fd; + + *retsock = NULL; + + memset(&addr, 0, sizeof(addr)); + + addr.len = sizeof(addr.data.un); + + if ((fd = socket(PF_UNIX, SOCK_STREAM, 0)) < 0) { + virReportSystemError(errno, "%s", _("Failed to create socket")); + goto error; + } + + addr.data.un.sun_family = AF_UNIX; + if (virStrcpyStatic(addr.data.un.sun_path, path) == NULL) { + virReportSystemError(ENOMEM, _("Path %s too long for unix socket"), path); + goto error; + } + if (addr.data.un.sun_path[0] == '@') + addr.data.un.sun_path[0] = '\0'; + else + unlink(addr.data.un.sun_path); + + oldgrp = getgid(); + oldmask = umask(~mask); + if (grp != 0 && setgid(grp) < 0) { + virReportSystemError(errno, + _("Failed to set group ID to %d"), grp); + goto error; + } + + if (bind(fd, &addr.data.sa, addr.len) < 0) { + virReportSystemError(errno, + _("Failed to bind socket to '%s'"), + path); + goto error; + } + umask(oldmask); + if (grp != 0 && setgid(oldgrp)) { + virReportSystemError(errno, + _("Failed to restore group ID to %d"), oldgrp); + goto error; + } + + if (!(*retsock = virNetSocketNew(&addr, NULL, fd, -1, 0))) + goto error; + + return 0; + +error: + VIR_FORCE_CLOSE(fd); + return -1; +} + + +int virNetSocketNewConnectTCP(const char *nodename, + const char *service, + virNetSocketPtr *retsock) +{ + virNetSocketPtr *socks = NULL; + size_t nsocks = 0; + struct addrinfo *ai; + struct addrinfo hints; + int fd = -1; + virSocketAddr localAddr; + virSocketAddr remoteAddr; + struct addrinfo *runp; + int savedErrno = ENOENT; + + *retsock = NULL; + + memset(&localAddr, 0, sizeof(localAddr)); + memset(&remoteAddr, 0, sizeof(remoteAddr)); + + memset (&hints, 0, sizeof hints); + hints.ai_flags = AI_PASSIVE | AI_ADDRCONFIG; + hints.ai_socktype = SOCK_STREAM; + + int e = getaddrinfo (nodename, service, &hints, &ai); + if (e != 0) { + virNetError(VIR_ERR_SYSTEM_ERROR, + _("Unable to resolve address '%s' service '%s': %s"), + nodename, service, gai_strerror (e)); + goto error; + } + + runp = ai; + while (runp) { + int opt = 1; + + if ((fd = socket(runp->ai_family, runp->ai_socktype, + runp->ai_protocol)) < 0) { + virReportSystemError(errno, "%s", _("Unable to create socket")); + goto error; + } + + setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof opt); + + if (connect(fd, runp->ai_addr, runp->ai_addrlen) >= 0) + break; + + savedErrno = errno; + VIR_FORCE_CLOSE(fd); + runp = runp->ai_next; + } + + if (fd == -1) { + virReportSystemError(savedErrno, + _("unable to connect to server at '%s:%s'"), + nodename, service); + return -1; + } + + freeaddrinfo (ai); + + if (getsockname(fd, &localAddr.data.sa, &localAddr.len) < 0) { + virReportSystemError(errno, "%s", _("Unable to get local socket name")); + goto error; + } + + if (getpeername(fd, &remoteAddr.data.sa, &remoteAddr.len) < 0) { + virReportSystemError(errno, "%s", _("Unable to get remote socket name")); + goto error; + } + + if (VIR_EXPAND_N(socks, nsocks, 1) < 0) { + virReportOOMError(); + goto error; + } + + if (!(*retsock = virNetSocketNew(&localAddr, &remoteAddr, fd, -1, 0))) + goto error; + + return 0; + +error: + VIR_FORCE_CLOSE(fd); + return -1; +} + + +int virNetSocketNewConnectUNIX(const char *path, + bool spawnDaemon, + const char *binary, + virNetSocketPtr *retsock) +{ + virSocketAddr localAddr; + virSocketAddr remoteAddr; + int fd; + int retries = 0; + + memset(&localAddr, 0, sizeof(localAddr)); + memset(&remoteAddr, 0, sizeof(remoteAddr)); + + remoteAddr.len = sizeof(remoteAddr.data.un); + + if ((fd = socket(PF_UNIX, SOCK_STREAM, 0)) < 0) { + virReportSystemError(errno, "%s", _("Failed to create socket")); + goto error; + } + + remoteAddr.data.un.sun_family = AF_UNIX; + if (virStrcpyStatic(remoteAddr.data.un.sun_path, path) == NULL) { + virReportSystemError(ENOMEM, _("Path %s too long for unix socket"), path); + goto error; + } + if (remoteAddr.data.un.sun_path[0] == '@') + remoteAddr.data.un.sun_path[0] = '\0'; + else + unlink(remoteAddr.data.un.sun_path); + +retry: + if (connect(fd, &remoteAddr.data.sa, remoteAddr.len) < 0) { + if (errno == ECONNREFUSED && spawnDaemon && retries < 20) { + if (retries == 0 && + virNetSocketForkDaemon(binary) < 0) + goto error; + + retries++; + usleep(1000 * 100 * retries); + goto retry; + } + + virReportSystemError(errno, + _("Failed to connect socket to '%s'"), + path); + goto error; + } + +#if 0 + /* There is no meaningful local addr for UNIX sockets, + * and getsockname() returns something with AF_INET + * in sa_family when run against AF_JUNIX sockets ! + */ + if (getsockname(fd, &localAddr.data.sa, &localAddr.len) < 0) { + virReportSystemError(errno, "%s", _("Unable to get local socket name")); + goto error; + } +#else + localAddr.data.sa.sa_family = AF_UNIX; +#endif + + VIR_WARN("%d %d", localAddr.len, localAddr.data.sa.sa_family); + + if (!(*retsock = virNetSocketNew(&localAddr, &remoteAddr, fd, -1, 0))) + goto error; + + return 0; + +error: + VIR_FORCE_CLOSE(fd); + return -1; +} + +int virNetSocketNewConnectSSH(const char *nodename, + const char *service, + const char *binary, + const char *username, + bool noTTY, + const char *netcat, + const char *path, + virNetSocketPtr *retsock) +{ + const char **cmdargv = NULL; + int ncmdargv; + + *retsock = NULL; + + ncmdargv = 6; + if (username) ncmdargv += 2; /* For -l username */ + if (noTTY) ncmdargv += 5; /* For -T -o BatchMode=yes -e none */ + if (service) ncmdargv += 2; /* For -p port */ + + /* + * Generate the final command argv[] array. + * ssh [-p $port] [-l $username] $hostname $netcat -U $sockname [NULL] + */ + if (VIR_ALLOC_N(cmdargv, ncmdargv) < 0) + goto no_memory; + + ncmdargv = 0; + cmdargv[ncmdargv++] = binary; + if (service) { + cmdargv[ncmdargv++] = "-p"; + cmdargv[ncmdargv++] = service; + } + if (username) { + cmdargv[ncmdargv++] = "-l"; + cmdargv[ncmdargv++] = username; + } + if (noTTY) { + cmdargv[ncmdargv++] = "-T"; + cmdargv[ncmdargv++] = "-o"; + cmdargv[ncmdargv++] = "BatchMode=yes"; + cmdargv[ncmdargv++] = "-e"; + cmdargv[ncmdargv++] = "none"; + } + cmdargv[ncmdargv++] = nodename; + cmdargv[ncmdargv++] = netcat ? netcat : "nc"; + cmdargv[ncmdargv++] = "-U"; + cmdargv[ncmdargv++] = path; + cmdargv[ncmdargv++] = NULL; + + return virNetSocketNewConnectCommand(cmdargv, NULL, retsock); + +no_memory: + VIR_FREE(cmdargv); + virReportOOMError(); + return -1; +} + + +int virNetSocketNewConnectCommand(const char **cmdargv, + const char **cmdenv, + virNetSocketPtr *retsock) +{ + pid_t pid = 0; + int sv[2]; + int errfd[2]; + + *retsock = NULL; + + /* Fork off the external process. Use socketpair to create a private + * (unnamed) Unix domain socket to the child process so we don't have + * to faff around with two file descriptors (a la 'pipe(2)'). + */ + if (socketpair(PF_UNIX, SOCK_STREAM, 0, sv) < 0) { + virReportSystemError(errno, "%s", + _("unable to create socket pair")); + goto error; + } + + if (pipe(errfd) < 0) { + virReportSystemError(errno, "%s", + _("unable to create socket pair")); + goto error; + } + + if (virExec(cmdargv, cmdenv, NULL, + &pid, sv[1], &(sv[1]), &(errfd[1]), + VIR_EXEC_CLEAR_CAPS) < 0) + goto error; + + /* Parent continues here. */ + VIR_FORCE_CLOSE(sv[1]); + VIR_FORCE_CLOSE(errfd[1]); + + if (!(*retsock = virNetSocketNew(NULL, NULL, sv[0], errfd[0], pid))) + goto error; + + return 0; + +error: + VIR_FORCE_CLOSE(sv[0]); + VIR_FORCE_CLOSE(sv[1]); + VIR_FORCE_CLOSE(errfd[1]); + VIR_FORCE_CLOSE(errfd[1]); + + if (pid > 0) { + pid_t reap; + do { +retry: + reap = waitpid(pid, NULL, 0); + if (reap == -1 && errno == EINTR) + goto retry; + } while (reap != -1 && reap != pid); + } + + return -1; +} + + +void virNetSocketFree(virNetSocketPtr sock) +{ + VIR_DEBUG("sock=%p", sock); + + if (!sock) + return; + + if (sock->watch) { + virEventRemoveHandle(sock->watch); + sock->watch = -1; + } + + if (sock->localAddr.data.sa.sa_family == AF_UNIX && + sock->localAddr.data.un.sun_path[0] != '\0') + unlink(sock->localAddr.data.un.sun_path); + + VIR_FORCE_CLOSE(sock->fd); + VIR_FORCE_CLOSE(sock->errfd); + + if (sock->pid > 0) { + pid_t reap; + kill(sock->pid, SIGTERM); + do { +retry: + reap = waitpid(sock->pid, NULL, 0); + if (reap == -1 && errno == EINTR) + goto retry; + } while (reap != -1 && reap != sock->pid); + } + + VIR_FREE(sock->localAddrStr); + VIR_FREE(sock->remoteAddrStr); + + VIR_FREE(sock); +} + + +int virNetSocketFD(virNetSocketPtr sock) +{ + return sock->fd; +} + + +const char *virNetSocketLocalAddrString(virNetSocketPtr sock) +{ + return sock->localAddrStr; +} + +const char *virNetSocketRemoteAddrString(virNetSocketPtr sock) +{ + return sock->remoteAddrStr; +} + +ssize_t virNetSocketRead(virNetSocketPtr sock, char *buf, size_t len) +{ + return read(sock->fd, buf, len); +} + +ssize_t virNetSocketWrite(virNetSocketPtr sock, const char *buf, size_t len) +{ + return write(sock->fd, buf, len); +} + + +int virNetSocketListen(virNetSocketPtr sock) +{ + if (listen(sock->fd, 30) < 0) { + virReportSystemError(errno, "%s", _("Unable to listen on socket")); + return -1; + } + return 0; +} + +int virNetSocketAccept(virNetSocketPtr sock, virNetSocketPtr *clientsock) +{ + int fd; + virSocketAddr localAddr; + virSocketAddr remoteAddr; + + *clientsock = NULL; + + memset(&localAddr, 0, sizeof(localAddr)); + memset(&remoteAddr, 0, sizeof(remoteAddr)); + + remoteAddr.len = sizeof(remoteAddr.data.stor); + if ((fd = accept(sock->fd, &remoteAddr.data.sa, &remoteAddr.len)) < 0) { + if (errno == ECONNABORTED || + errno == EAGAIN) + return 0; + + virReportSystemError(errno, "%s", + _("Unable to accept client")); + return -1; + } + + if (getsockname(fd, &localAddr.data.sa, &localAddr.len) < 0) { + virReportSystemError(errno, "%s", _("Unable to get local socket name")); + VIR_FORCE_CLOSE(fd); + return -1; + } + + + if (!(*clientsock = virNetSocketNew(&localAddr, &remoteAddr, fd, -1, 0))) + return -1; + + return 0; +} + + +static void virNetSocketEventHandle(int fd ATTRIBUTE_UNUSED, + int watch ATTRIBUTE_UNUSED, + int events, + void *opaque) +{ + virNetSocketPtr sock = opaque; + + sock->func(sock, events, sock->opaque); +} + +int virNetSocketAddIOCallback(virNetSocketPtr sock, + int events, + virNetSocketIOFunc func, + void *opaque) +{ + if (sock->watch) { + VIR_DEBUG("Watch already registered on socket %p", sock); + return -1; + } + + if ((sock->watch = virEventAddHandle(sock->fd, + events, + virNetSocketEventHandle, + sock, + NULL)) < 0) { + VIR_WARN("Failed to register watch on socket %p", sock); + return -1; + } + sock->func = func; + sock->opaque = opaque; + + return 0; +} + +void virNetSocketUpdateIOCallback(virNetSocketPtr sock, + int events) +{ + if (!sock->watch) { + VIR_DEBUG("Watch not registered on socket %p", sock); + return; + } + + virEventUpdateHandle(sock->watch, events); +} + +void virNetSocketRemoveIOCallback(virNetSocketPtr sock) +{ + if (!sock->watch) { + VIR_DEBUG("Watch not registered on socket %p", sock); + return; + } + + virEventRemoveHandle(sock->watch); +} + diff --git a/src/rpc/virnetsocket.h b/src/rpc/virnetsocket.h new file mode 100644 index 0000000..a25918d --- /dev/null +++ b/src/rpc/virnetsocket.h @@ -0,0 +1,97 @@ +/* + * virnetsocket.h: generic network socket handling + * + * Copyright (C) 2006-2010 Red Hat, Inc. + * Copyright (C) 2006 Daniel P. Berrange + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA + * + * Author: Daniel P. Berrange <berrange@redhat.com> + */ + +#ifndef __VIR_NET_SOCKET_H__ +#define __VIR_NET_SOCKET_H__ + +#include "network.h" + +typedef struct _virNetSocket virNetSocket; +typedef virNetSocket *virNetSocketPtr; + + +typedef void (*virNetSocketIOFunc)(virNetSocketPtr sock, + int events, + void *opaque); + + +int virNetSocketNewListenTCP(const char *nodename, + const char *service, + virNetSocketPtr **addrs, + size_t *naddrs); + +int virNetSocketNewListenUNIX(const char *path, + mode_t mask, + gid_t grp, + virNetSocketPtr *addr); + +int virNetSocketNewConnectTCP(const char *nodename, + const char *service, + virNetSocketPtr *addr); + +int virNetSocketNewConnectUNIX(const char *path, + bool spawnDaemon, + const char *binary, + virNetSocketPtr *addr); + +int virNetSocketNewConnectSSH(const char *nodename, + const char *service, + const char *binary, + const char *username, + bool noTTY, + const char *netcat, + const char *path, + virNetSocketPtr *addr); + +int virNetSocketNewConnectCommand(const char **cmdargv, + const char **cmdenv, + virNetSocketPtr *addr); + +/* XXX bad */ +int virNetSocketFD(virNetSocketPtr sock); + +ssize_t virNetSocketRead(virNetSocketPtr sock, char *buf, size_t len); +ssize_t virNetSocketWrite(virNetSocketPtr sock, const char *buf, size_t len); + +void virNetSocketFree(virNetSocketPtr sock); + +const char *virNetSocketLocalAddrString(virNetSocketPtr sock); +const char *virNetSocketRemoteAddrString(virNetSocketPtr sock); + +int virNetSocketListen(virNetSocketPtr sock); +int virNetSocketAccept(virNetSocketPtr sock, + virNetSocketPtr *clientsock); + +int virNetSocketAddIOCallback(virNetSocketPtr sock, + int events, + virNetSocketIOFunc func, + void *opaque); + +void virNetSocketUpdateIOCallback(virNetSocketPtr sock, + int events); + +void virNetSocketRemoveIOCallback(virNetSocketPtr sock); + + + +#endif /* __VIR_NET_SOCKET_H__ */ diff --git a/src/rpc/virnettlscontext.c b/src/rpc/virnettlscontext.c new file mode 100644 index 0000000..9690328 --- /dev/null +++ b/src/rpc/virnettlscontext.c @@ -0,0 +1,611 @@ + + +#include <config.h> + +#include <unistd.h> +#include <fnmatch.h> +#include <stdlib.h> + +#include <gnutls/gnutls.h> +#include <gnutls/x509.h> +#include "gnutls_1_0_compat.h" + +#include "virnettlscontext.h" + +#include "memory.h" +#include "virterror_internal.h" +#include "util.h" +#include "logging.h" + +#define DH_BITS 1024 + +#define VIR_FROM_THIS VIR_FROM_RPC + +#define virNetError(code, ...) \ + virReportErrorHelper(NULL, VIR_FROM_RPC, code, __FILE__, \ + __FUNCTION__, __LINE__, __VA_ARGS__) + +struct _virNetTLSContext { + int refs; + + gnutls_certificate_credentials_t x509cred; + gnutls_dh_params_t dhParams; + + bool isServer; + bool requireValidCert; + const char *const*x509dnWhitelist; +}; + +struct _virNetTLSSession { + int refs; + + char *hostname; + gnutls_session_t session; + virNetTLSSessionWriteFunc writeFunc; + virNetTLSSessionReadFunc readFunc; + void *opaque; +}; + + +static int +virNetTLSContextCheckCertFile(const char *type, const char *file) +{ + if (access(file, R_OK) < 0) { + virReportSystemError(errno, + _("Cannot read %s '%s'"), + type, file); + return -1; + } + return 0; +} + + +static void virNetTLSLog(int level, const char *str) { + VIR_DEBUG("%d %s", level, str); +} + +static virNetTLSContextPtr virNetTLSContextNew(const char *ca_file, + const char *crl_file, + const char *cert_file, + const char *key_file, + const char *const*x509dnWhitelist, + bool requireValidCert, + bool isServer) +{ + virNetTLSContextPtr ctxt; + char *gnutlsdebug; + int err; + + if (VIR_ALLOC(ctxt) < 0) { + virReportOOMError(); + return NULL; + } + + ctxt->refs = 1; + + /* Initialise GnuTLS. */ + gnutls_global_init(); + + if ((gnutlsdebug = getenv("LIBVIRT_GNUTLS_DEBUG")) != NULL) { + int val; + if (virStrToLong_i(gnutlsdebug, NULL, 10, &val) < 0) + val = 10; + gnutls_global_set_log_level(val); + gnutls_global_set_log_function(virNetTLSLog); + } + + + err = gnutls_certificate_allocate_credentials(&ctxt->x509cred); + if (err) { + virNetError(VIR_ERR_SYSTEM_ERROR, + _("Unable to allocate x509 credentials: %s"), + gnutls_strerror (err)); + goto error; + } + + if (ca_file && ca_file[0] != '\0') { + if (virNetTLSContextCheckCertFile("CA certificate", ca_file) < 0) + goto error; + + VIR_DEBUG("loading CA cert from %s", ca_file); + err = gnutls_certificate_set_x509_trust_file(ctxt->x509cred, + ca_file, + GNUTLS_X509_FMT_PEM); + if (err < 0) { + virNetError(VIR_ERR_SYSTEM_ERROR, + _("Unable to set x509 CA certificate: %s"), + gnutls_strerror (err)); + goto error; + } + } + + if (crl_file && crl_file[0] != '\0') { + if (virNetTLSContextCheckCertFile("CA revocation list", crl_file) < 0) + goto error; + + VIR_DEBUG("loading CRL from %s", crl_file); + err = gnutls_certificate_set_x509_crl_file(ctxt->x509cred, + crl_file, + GNUTLS_X509_FMT_PEM); + if (err < 0) { + virNetError(VIR_ERR_SYSTEM_ERROR, + _("Unable to set x509 certificate revocation list: %s"), + gnutls_strerror (err)); + goto error; + } + } + + if (cert_file && cert_file[0] != '\0' && key_file && key_file[0] != '\0') { + if (virNetTLSContextCheckCertFile("server certificate", cert_file) < 0) + goto error; + if (virNetTLSContextCheckCertFile("server key", key_file) < 0) + goto error; + VIR_DEBUG("loading cert and key from %s and %s", cert_file, key_file); + err = + gnutls_certificate_set_x509_key_file(ctxt->x509cred, + cert_file, key_file, + GNUTLS_X509_FMT_PEM); + if (err < 0) { + virNetError(VIR_ERR_SYSTEM_ERROR, + _("Unable to set x509 key and certificate: %s"), + gnutls_strerror (err)); + goto error; + } + } + + /* Generate Diffie Hellman parameters - for use with DHE + * kx algorithms. These should be discarded and regenerated + * once a day, once a week or once a month. Depending on the + * security requirements. + */ + if (isServer) { + err = gnutls_dh_params_init(&ctxt->dhParams); + if (err < 0) { + virNetError(VIR_ERR_SYSTEM_ERROR, + _("Unable to initialize diffie-hellman parameters: %s"), + gnutls_strerror (err)); + goto error; + } + err = gnutls_dh_params_generate2(ctxt->dhParams, DH_BITS); + if (err < 0) { + virNetError(VIR_ERR_SYSTEM_ERROR, + _("Unable to generate diffie-hellman parameters: %s"), + gnutls_strerror (err)); + goto error; + } + + gnutls_certificate_set_dh_params(ctxt->x509cred, + ctxt->dhParams); + } + + ctxt->requireValidCert = requireValidCert; + ctxt->x509dnWhitelist = x509dnWhitelist; + ctxt->isServer = isServer; + + return ctxt; + +error: + if (isServer) + gnutls_dh_params_deinit(ctxt->dhParams); + gnutls_certificate_free_credentials(ctxt->x509cred); + VIR_FREE(ctxt); + return NULL; +} + + +virNetTLSContextPtr virNetTLSContextNewServer(const char *ca_file, + const char *crl_file, + const char *cert_file, + const char *key_file, + const char *const*x509dnWhitelist, + bool requireValidCert) +{ + return virNetTLSContextNew(ca_file, crl_file, cert_file, key_file, + x509dnWhitelist, requireValidCert, true); +} + +virNetTLSContextPtr virNetTLSContextNewClient(const char *ca_file, + const char *cert_file, + const char *key_file, + bool requireValidCert) +{ + return virNetTLSContextNew(ca_file, NULL, cert_file, key_file, + NULL, requireValidCert, false); +} + + +void virNetTLSContextRef(virNetTLSContextPtr ctxt) +{ + ctxt->refs++; +} + + +/* Check DN is on tls_allowed_dn_list. */ +static int +virNetTLSContextCheckDN(virNetTLSContextPtr ctxt, + const char *dname) +{ + const char *const*wildcards; + + /* If the list is not set, allow any DN. */ + wildcards = ctxt->x509dnWhitelist; + if (!wildcards) + return 1; + + while (*wildcards) { + if (fnmatch (*wildcards, dname, 0) == 0) + return 1; + wildcards++; + } + + /* Print the client's DN. */ + DEBUG(_("Failed whitelist check for client DN '%s'"), dname); + + return 0; // Not found. +} + +static int virNetTLSContextValidCertificate(virNetTLSContextPtr ctxt, + virNetTLSSessionPtr sess) +{ + int ret; + unsigned int status; + const gnutls_datum_t *certs; + unsigned int nCerts, i; + time_t now; + char name[256]; + size_t namesize = sizeof name; + + memset(name, 0, namesize); + + if ((ret = gnutls_certificate_verify_peers2(sess->session, &status)) < 0){ + virNetError(VIR_ERR_SYSTEM_ERROR, + _("Unable to verify TLS peer: %s"), + gnutls_strerror(ret)); + goto authdeny; + } + + if ((now = time(NULL)) == ((time_t)-1)) { + virReportSystemError(errno, "%s", + _("cannot get current time")); + goto authfail; + } + + if (status != 0) { + const char *reason = _("Invalid certificate"); + + if (status & GNUTLS_CERT_INVALID) + reason = _("The certificate is not trusted."); + + if (status & GNUTLS_CERT_SIGNER_NOT_FOUND) + reason = _("The certificate hasn't got a known issuer."); + + if (status & GNUTLS_CERT_REVOKED) + reason = _("The certificate has been revoked."); + +#ifndef GNUTLS_1_0_COMPAT + if (status & GNUTLS_CERT_INSECURE_ALGORITHM) + reason = _("The certificate uses an insecure algorithm"); +#endif + + virNetError(VIR_ERR_SYSTEM_ERROR, + _("Certificate failed validation: %s"), + reason); + goto authdeny; + } + + if (gnutls_certificate_type_get(sess->session) != GNUTLS_CRT_X509) { + virNetError(VIR_ERR_SYSTEM_ERROR, "%s", + _("Only x509 certificates are supported")); + goto authdeny; + } + + if (!(certs = gnutls_certificate_get_peers(sess->session, &nCerts))) { + virNetError(VIR_ERR_SYSTEM_ERROR, "%s", + _("The certificate has no peers")); + goto authdeny; + } + + for (i = 0; i < nCerts; i++) { + gnutls_x509_crt_t cert; + + if (gnutls_x509_crt_init (&cert) < 0) { + virNetError(VIR_ERR_SYSTEM_ERROR, "%s", + _("Unable to initialize certificate")); + goto authfail; + } + + if (gnutls_x509_crt_import(cert, &certs[i], GNUTLS_X509_FMT_DER) < 0) { + virNetError(VIR_ERR_SYSTEM_ERROR, "%s", + _("Unable to load certificate")); + gnutls_x509_crt_deinit(cert); + goto authfail; + } + + if (gnutls_x509_crt_get_expiration_time(cert) < now) { + virNetError(VIR_ERR_SYSTEM_ERROR, "%s", + _("The client certificate has expired")); + gnutls_x509_crt_deinit(cert); + goto authdeny; + } + + if (gnutls_x509_crt_get_activation_time(cert) > now) { + virNetError(VIR_ERR_SYSTEM_ERROR, "%s", + _("The client certificate is not yet active")); + gnutls_x509_crt_deinit(cert); + goto authdeny; + } + + if (i == 0) { + ret = gnutls_x509_crt_get_dn(cert, name, &namesize); + if (ret != 0) { + virNetError(VIR_ERR_SYSTEM_ERROR, + _("Failed to get certificate distinguished name: %s"), + gnutls_strerror(ret)); + gnutls_x509_crt_deinit(cert); + goto authfail; + } + + if (!virNetTLSContextCheckDN(ctxt, name)) { + /* This is the most common error: make it informative. */ + virNetError(VIR_ERR_SYSTEM_ERROR, "%s", + _("Client's Distinguished Name is not on the list " + "of allowed clients (tls_allowed_dn_list). Use " + "'certtool -i --infile clientcert.pem' to view the" + "Distinguished Name field in the client certificate," + "or run this daemon with --verbose option.")); + gnutls_x509_crt_deinit(cert); + goto authdeny; + } + + if (sess->hostname && + !gnutls_x509_crt_check_hostname(cert, sess->hostname)) { + virNetError(VIR_ERR_RPC, + _("Certificate's owner does not match the hostname (%s)"), + sess->hostname); + gnutls_x509_crt_deinit(cert); + goto authdeny; + } + } + } + +#if 0 + PROBE(CLIENT_TLS_ALLOW, "fd=%d, name=%s", client->fd, (char *)name); +#endif + return 0; + +authdeny: +#if 0 + PROBE(CLIENT_TLS_DENY, "fd=%d, name=%s", client->fd, (char *)name); +#endif + return -1; + +authfail: +#if 0 + PROBE(CLIENT_TLS_FAIL, "fd=%d", client->fd); +#endif + return -1; +} + +int virNetTLSContextCheckCertificate(virNetTLSContextPtr ctxt, + virNetTLSSessionPtr sess) { + if (virNetTLSContextValidCertificate(ctxt, sess) < 0) { + if (ctxt->requireValidCert) { + virNetError(VIR_ERR_AUTH_FAILED, "%s", + _("Failed to verify peer's certificate")); + return -1; + } + VIR_INFO0(_("Ignoring bad certificate at user request")); + } + return 0; +} + +void virNetTLSContextFree(virNetTLSContextPtr ctxt) +{ + if (!ctxt) + return; + + ctxt->refs--; + if (ctxt->refs > 0) + return; + + gnutls_dh_params_deinit(ctxt->dhParams); + gnutls_certificate_free_credentials(ctxt->x509cred); + VIR_FREE(ctxt); +} + + + +static ssize_t +virNetTLSSessionPush(void *opaque, const void *buf, size_t len) +{ + virNetTLSSessionPtr sess = opaque; + return sess->writeFunc(buf, len, sess->opaque); +} + + +static ssize_t +virNetTLSSessionPull(void *opaque, void *buf, size_t len) +{ + virNetTLSSessionPtr sess = opaque; + return sess->readFunc(buf, len, sess->opaque); +} + + +virNetTLSSessionPtr virNetTLSSessionNew(virNetTLSContextPtr ctxt, + const char *hostname, + virNetTLSSessionWriteFunc writeFunc, + virNetTLSSessionReadFunc readFunc, + void *opaque) +{ + virNetTLSSessionPtr sess; + int err; + static const int cert_type_priority[] = { GNUTLS_CRT_X509, 0 }; + + if (VIR_ALLOC(sess) < 0) { + virReportOOMError(); + return NULL; + } + + sess->refs = 1; + sess->writeFunc = writeFunc; + sess->readFunc = readFunc; + sess->opaque = opaque; + if (!(sess->hostname = strdup(hostname))) { + virReportOOMError(); + goto error; + } + + if ((err = gnutls_init(&sess->session, + ctxt->isServer ? GNUTLS_SERVER : GNUTLS_CLIENT)) != 0) { + virNetError(VIR_ERR_SYSTEM_ERROR, + _("Failed to initialize TLS session: %s"), + gnutls_strerror(err)); + goto error; + } + + /* avoid calling all the priority functions, since the defaults + * are adequate. + */ + if ((err = gnutls_set_default_priority(sess->session)) != 0 || + (err = gnutls_certificate_type_set_priority(sess->session, + cert_type_priority))) { + virNetError(VIR_ERR_SYSTEM_ERROR, + _("Failed to set TLS session priority %s"), + gnutls_strerror(err)); + goto error; + } + + if ((err = gnutls_credentials_set(sess->session, + GNUTLS_CRD_CERTIFICATE, + ctxt->x509cred)) != 0) { + virNetError(VIR_ERR_SYSTEM_ERROR, + _("Failed set TLS x509 credentials: %s"), + gnutls_strerror(err)); + goto error; + } + + /* request client certificate if any. + */ + if (ctxt->isServer) { + gnutls_certificate_server_set_request(sess->session, GNUTLS_CERT_REQUEST); + + gnutls_dh_set_prime_bits(sess->session, DH_BITS); + } + + gnutls_transport_set_ptr(sess->session, sess); + gnutls_transport_set_push_function(sess->session, + virNetTLSSessionPush); + gnutls_transport_set_pull_function(sess->session, + virNetTLSSessionPull); + + return sess; + +error: + virNetTLSSessionFree(sess); + return NULL; +} + + +void virNetTLSSessionRef(virNetTLSSessionPtr sess) +{ + sess->refs++; +} + + +ssize_t virNetTLSSessionWrite(virNetTLSSessionPtr sess, + const char *buf, size_t len) +{ + int ret; + ret = gnutls_record_send(sess->session, buf, len); + + switch (ret) { + case GNUTLS_E_AGAIN: + errno = EAGAIN; + break; + case GNUTLS_E_INTERRUPTED: + errno = EINTR; + break; + case 0: + break; + default: + errno = EIO; + } + + return ret >= 0 ? ret : -1; +} + +ssize_t virNetTLSSessionRead(virNetTLSSessionPtr sess, + char *buf, size_t len) +{ + int ret; + + ret = gnutls_record_recv(sess->session, buf, len); + + switch (ret) { + case GNUTLS_E_AGAIN: + errno = EAGAIN; + break; + case GNUTLS_E_INTERRUPTED: + errno = EINTR; + break; + case 0: + break; + default: + errno = EIO; + } + + return ret >= 0 ? ret : -1; +} + +int virNetTLSSessionHandshake(virNetTLSSessionPtr sess) +{ + int ret = gnutls_handshake(sess->session); + + if (ret == 0) + return 0; + if (ret == GNUTLS_E_INTERRUPTED || ret == GNUTLS_E_AGAIN) + return 1; + + virNetError(VIR_ERR_AUTH_FAILED, + _("TLS handshake failed %s"), + gnutls_strerror (ret)); + return -1; +} + +int virNetTLSSessionHandshakeDirection(virNetTLSSessionPtr sess) +{ + if (gnutls_record_get_direction (sess->session) == 0) + return 0; + else + return 1; +} + +int virNetTLSSessionGetKeySize(virNetTLSSessionPtr sess) +{ + gnutls_cipher_algorithm_t cipher; + int ssf; + + cipher = gnutls_cipher_get(sess->session); + if (!(ssf = gnutls_cipher_get_key_size(cipher))) { + virNetError(VIR_ERR_INTERNAL_ERROR, "%s", + _("invalid cipher size for TLS session")); + return -1; + } + + return ssf; +} + + +void virNetTLSSessionFree(virNetTLSSessionPtr sess) +{ + if (!sess) + return; + + sess->refs--; + if (sess->refs > 0) + return; + + VIR_FREE(sess->hostname); + gnutls_deinit(sess->session); + VIR_FREE(sess); +} diff --git a/src/rpc/virnettlscontext.h b/src/rpc/virnettlscontext.h new file mode 100644 index 0000000..b93b379 --- /dev/null +++ b/src/rpc/virnettlscontext.h @@ -0,0 +1,63 @@ + + +#ifndef __VIR_NET_TLS_CONTEXT_H__ +#define __VIR_NET_TLS_CONTEXT_H__ + +#include <stdbool.h> +#include <sys/types.h> + +typedef struct _virNetTLSContext virNetTLSContext; +typedef virNetTLSContext *virNetTLSContextPtr; + +typedef struct _virNetTLSSession virNetTLSSession; +typedef virNetTLSSession *virNetTLSSessionPtr; + + +virNetTLSContextPtr virNetTLSContextNewServer(const char *ca_file, + const char *crl_file, + const char *cert_file, + const char *key_file, + const char *const*x509dnWhitelist, + bool requireValidCert); + +virNetTLSContextPtr virNetTLSContextNewClient(const char *ca_file, + const char *cert_file, + const char *key_file, + bool requireValidCert); + +void virNetTLSContextRef(virNetTLSContextPtr ctxt); + +int virNetTLSContextCheckCertificate(virNetTLSContextPtr ctxt, + virNetTLSSessionPtr sess); + +void virNetTLSContextFree(virNetTLSContextPtr ctxt); + + +typedef ssize_t (*virNetTLSSessionWriteFunc)(const char *buf, size_t len, + void *opaque); +typedef ssize_t (*virNetTLSSessionReadFunc)(char *buf, size_t len, + void *opaque); + +virNetTLSSessionPtr virNetTLSSessionNew(virNetTLSContextPtr ctxt, + const char *hostname, + virNetTLSSessionWriteFunc writeFunc, + virNetTLSSessionReadFunc readFunc, + void *opaque); + +void virNetTLSSessionRef(virNetTLSSessionPtr sess); + +ssize_t virNetTLSSessionWrite(virNetTLSSessionPtr sess, + const char *buf, size_t len); +ssize_t virNetTLSSessionRead(virNetTLSSessionPtr sess, + char *buf, size_t len); + +int virNetTLSSessionHandshake(virNetTLSSessionPtr sess); + +int virNetTLSSessionHandshakeDirection(virNetTLSSessionPtr sess); + +int virNetTLSSessionGetKeySize(virNetTLSSessionPtr sess); + +void virNetTLSSessionFree(virNetTLSSessionPtr sess); + + +#endif -- 1.7.2.3

To facilitate creation of new daemons providing XDR RPC services, pull alot of the libvirtd daemon code into a set of reusable objects. * virNetServer: A server contains one or more services which accept incoming clients. It maintains the list of active clients. It has a list of RPC programs which can be used by clients. When clients produce a complete RPC message, the server passes this onto the corresponding program for handling, and queues any response back with the client. * virNetServerClient: Encapsulates a single client connection. All I/O for the client is handled, reading & writing RPC messages. Also contains the SASL/TLS code, but this will eventually move into the virNetSocket object * virNetServerProgram: Handles processing and dispatch of RPC method calls for a single RPC (program,version). Multiple programs can be registered with the server. * virNetServerService: Encapsulates socket(s) listening for new connections. Each service listens on a single host/port, but may have multiple sockets if on a dual IPv4/6 host. Each new daemon now merely has to define the list of RPC procedures & their handlers. It does not need to deal with any network related functionality at all. --- src/Makefile.am | 18 +- src/rpc/virnetserver.c | 654 +++++++++++++++++++++++++++ src/rpc/virnetserver.h | 74 ++++ src/rpc/virnetserverclient.c | 974 +++++++++++++++++++++++++++++++++++++++++ src/rpc/virnetserverclient.h | 40 ++ src/rpc/virnetservermessage.h | 20 + src/rpc/virnetserverprogram.c | 437 ++++++++++++++++++ src/rpc/virnetserverprogram.h | 76 ++++ src/rpc/virnetserverservice.c | 208 +++++++++ src/rpc/virnetserverservice.h | 32 ++ 10 files changed, 2532 insertions(+), 1 deletions(-) create mode 100644 src/rpc/virnetserver.c create mode 100644 src/rpc/virnetserver.h create mode 100644 src/rpc/virnetserverclient.c create mode 100644 src/rpc/virnetserverclient.h create mode 100644 src/rpc/virnetservermessage.h create mode 100644 src/rpc/virnetserverprogram.c create mode 100644 src/rpc/virnetserverprogram.h create mode 100644 src/rpc/virnetserverservice.c create mode 100644 src/rpc/virnetserverservice.h diff --git a/src/Makefile.am b/src/Makefile.am index 613ff0a..e78a0af 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -1117,7 +1117,7 @@ libvirt_qemu_la_LIBADD = libvirt.la $(CYGWIN_EXTRA_LIBADD) EXTRA_DIST += $(LIBVIRT_QEMU_SYMBOL_FILE) -noinst_LTLIBRARIES += libvirt-net-rpc.la +noinst_LTLIBRARIES += libvirt-net-rpc.la libvirt-net-rpc-server.la libvirt_net_rpc_la_SOURCES = \ ../daemon/event.c \ @@ -1138,6 +1138,22 @@ libvirt_net_rpc_la_LDFLAGS = \ libvirt_net_rpc_la_LIBADD = \ $(CYGWIN_EXTRA_LIBADD) +libvirt_net_server_la_SOURCES = \ + rpc/virnetservermessage.h \ + rpc/virnetserverprogram.h rpc/virnetserverprogram.c \ + rpc/virnetserverservice.h rpc/virnetserverservice.c \ + rpc/virnetserverclient.h rpc/virnetserverclient.c \ + rpc/virnetserver.h rpc/virnetserver.c +libvirt_net_server_la_CFLAGS = \ + $(AM_CFLAGS) +libvirt_net_server_la_LDFLAGS = \ + $(AM_LDFLAGS) \ + $(CYGWIN_EXTRA_LDFLAGS) \ + $(MINGW_EXTRA_LDFLAGS)l +libvirt_net_server_la_LIBADD = \ + $(CYGWIN_EXTRA_LIBADD) + + libexec_PROGRAMS = if WITH_STORAGE_DISK diff --git a/src/rpc/virnetserver.c b/src/rpc/virnetserver.c new file mode 100644 index 0000000..0384bb9 --- /dev/null +++ b/src/rpc/virnetserver.c @@ -0,0 +1,654 @@ +/* + * libvirtd.h: daemon data structure definitions + * + * Copyright (C) 2006-2010 Red Hat, Inc. + * Copyright (C) 2006 Daniel P. Berrange + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA + * + * Author: Daniel P. Berrange <berrange@redhat.com> + */ + +#include <config.h> + +#include <unistd.h> +#include <string.h> + +#include "virnetserver.h" +#include "logging.h" +#include "memory.h" +#include "virterror_internal.h" +#include "threads.h" +#include "threadpool.h" +#include "util.h" +#include "files.h" +#include "event.h" +#include "../daemon/event.h" + +#define VIR_FROM_THIS VIR_FROM_RPC + +#define virNetError(code, ...) \ + virReportErrorHelper(NULL, VIR_FROM_RPC, code, __FILE__, \ + __FUNCTION__, __LINE__, __VA_ARGS__) + +typedef struct _virNetServerSignal virNetServerSignal; +typedef virNetServerSignal *virNetServerSignalPtr; + +struct _virNetServerSignal { + struct sigaction oldaction; + int signum; + virNetServerSignalFunc func; + void *opaque; +}; + +typedef struct _virNetServerJob virNetServerJob; +typedef virNetServerJob *virNetServerJobPtr; + +struct _virNetServerJob { + virNetServerClientPtr client; + virNetServerMessagePtr msg; +}; + +struct _virNetServer { + int refs; + + virMutex lock; + + virThreadPoolPtr workers; + + size_t nsignals; + virNetServerSignalPtr *signals; + int sigread; + int sigwrite; + int sigwatch; + + size_t nservices; + virNetServerServicePtr *services; + + size_t nprograms; + virNetServerProgramPtr *programs; + + size_t nclients; + size_t nclients_alloc; + size_t nclients_max; + virNetServerClientPtr *clients; + + unsigned int quit :1; + + virNetTLSContextPtr tls; + + unsigned int autoShutdownTimeout; + virNetServerAutoShutdownFunc autoShutdownFunc; + void *autoShutdownOpaque; +}; + + +static void virNetServerLock(virNetServerPtr srv) +{ + virMutexLock(&srv->lock); +} + +static void virNetServerUnlock(virNetServerPtr srv) +{ + virMutexUnlock(&srv->lock); +} + + +static void virNetServerHandleJob(void *jobOpaque, void *opaque) +{ + virNetServerPtr srv = opaque; + virNetServerJobPtr job = jobOpaque; + virNetServerProgramPtr prog = NULL; + int i; + + virNetServerLock(srv); + VIR_DEBUG("server=%p client=%p message=%p", + srv, job->client, job->msg); + + for (i = 0 ; i < srv->nprograms ; i++) { + if (virNetServerProgramMatches(srv->programs[i], job->msg)) { + prog = srv->programs[i]; + break; + } + } + + if (!prog) { + VIR_DEBUG("Cannot find program %d version %d", + job->msg->msg.header.prog, + job->msg->msg.header.vers); + goto error; + } + + if (virNetServerProgramDispatch(prog, + srv, + job->client, + job->msg) < 0) { + job->msg = NULL; + goto error; + } + + VIR_FREE(job); + virNetServerUnlock(srv); + return; + +error: + if (job->msg) + virNetServerClientFinishMessage(job->client, job->msg); + virNetServerClientClose(job->client); + VIR_FREE(job); + virNetServerUnlock(srv); +} + + +static int virNetServerDispatchNewMessage(virNetServerClientPtr client, + virNetServerMessagePtr msg, + void *opaque) +{ + virNetServerPtr srv = opaque; + virNetServerJobPtr job; + + VIR_DEBUG("server=%p client=%p message=%p", + srv, client, msg); + + if (VIR_ALLOC(job) < 0) { + virNetServerClientFinishMessage(client, msg); + virNetServerClientClose(client); + virReportOOMError(); + return -1; + } + + job->client = client; + job->msg = msg; + + virNetServerLock(srv); + virThreadPoolSendJob(srv->workers, job); + virNetServerUnlock(srv); + + return 0; +} + + +static int virNetServerDispatchNewClient(virNetServerServicePtr svc ATTRIBUTE_UNUSED, + virNetServerClientPtr client, + void *opaque) +{ + virNetServerPtr srv = opaque; + + virNetServerLock(srv); + + if (srv->nclients >= srv->nclients_max) { + virNetError(VIR_ERR_RPC, + _("Too many active clients (%d), dropping connection from %s"), + (int)srv->nclients_max, virNetServerClientAddrString(client)); + goto error; + } + + if (virNetServerClientInit(client) < 0) + goto error; + + if (VIR_RESIZE_N(srv->clients, srv->nclients_alloc, + srv->nclients, 1) < 0) { + virReportOOMError(); + goto error; + } + srv->clients[srv->nclients++] = client; + virNetServerClientRef(client); + + virNetServerClientSetDispatcher(client, + virNetServerDispatchNewMessage, + srv); + + virNetServerUnlock(srv); + return 0; + +error: + virNetServerUnlock(srv); + return -1; +} + + +virNetServerPtr virNetServerNew(size_t min_workers, + size_t max_workers, + size_t max_clients) +{ + virNetServerPtr srv; + struct sigaction sig_action; + + if (VIR_ALLOC(srv) < 0) { + virReportOOMError(); + return NULL; + } + + srv->refs = 1; + + if (!(srv->workers = virThreadPoolNew(min_workers, max_workers, + virNetServerHandleJob, + srv))) + goto error; + + srv->nclients_max = max_clients; + srv->sigwrite = srv->sigread = -1; + + if (virMutexInit(&srv->lock) < 0) { + virNetError(VIR_ERR_INTERNAL_ERROR, "%s", + _("cannot initialize mutex")); + goto error; + } + + if (virEventInit() < 0) { + virNetError(VIR_ERR_INTERNAL_ERROR, "%s", + _("Failed to initialize event system")); + goto error; + } + + virEventRegisterImpl(virEventAddHandleImpl, + virEventUpdateHandleImpl, + virEventRemoveHandleImpl, + virEventAddTimeoutImpl, + virEventUpdateTimeoutImpl, + virEventRemoveTimeoutImpl); + + memset(&sig_action, 0, sizeof(sig_action)); + sig_action.sa_handler = SIG_IGN; + sigaction(SIGPIPE, &sig_action, NULL); + + return srv; + +error: + virNetServerFree(srv); + return NULL; +} + + +void virNetServerRef(virNetServerPtr srv) +{ + virNetServerLock(srv); + srv->refs++; + virNetServerUnlock(srv); +} + + +void virNetServerAutoShutdown(virNetServerPtr srv, + unsigned int timeout, + virNetServerAutoShutdownFunc func, + void *opaque) +{ + virNetServerLock(srv); + + srv->autoShutdownTimeout = timeout; + srv->autoShutdownFunc = func; + srv->autoShutdownOpaque = opaque; + + virNetServerUnlock(srv); +} + + +static sig_atomic_t sigErrors = 0; +static int sigLastErrno = 0; +static int sigWrite = -1; + +static void virNetServerSignalHandler(int sig, siginfo_t * siginfo, + void* context ATTRIBUTE_UNUSED) +{ + int origerrno; + int r; + + /* set the sig num in the struct */ + siginfo->si_signo = sig; + + origerrno = errno; + r = safewrite(sigWrite, siginfo, sizeof(*siginfo)); + if (r == -1) { + sigErrors++; + sigLastErrno = errno; + } + errno = origerrno; +} + +static void +virNetServerSignalEvent(int watch, + int fd ATTRIBUTE_UNUSED, + int events ATTRIBUTE_UNUSED, + void *opaque) { + virNetServerPtr srv = opaque; + siginfo_t siginfo; + int i; + + virNetServerLock(srv); + + if (saferead(srv->sigread, &siginfo, sizeof(siginfo)) != sizeof(siginfo)) { + virReportSystemError(errno, "%s", + _("Failed to read from signal pipe")); + virEventRemoveHandle(watch); + srv->sigwatch = -1; + goto cleanup; + } + + for (i = 0 ; i < srv->nsignals ; i++) { + if (siginfo.si_signo == srv->signals[i]->signum) { + virNetServerSignalFunc func = srv->signals[i]->func; + void *funcopaque = srv->signals[i]->opaque; + virNetServerUnlock(srv); + func(srv, &siginfo, funcopaque); + return; + } + } + + virNetError(VIR_ERR_INTERNAL_ERROR, + _("Unexpected signal received: %d"), siginfo.si_signo); + +cleanup: + virNetServerUnlock(srv); +} + +static int virNetServerSignalSetup(virNetServerPtr srv) +{ + int fds[2]; + + if (srv->sigwrite != -1) + return 0; + + if (pipe(fds) < 0) { + virReportSystemError(errno, "%s", + _("Unable to create signal pipe")); + return -1; + } + + if (virSetNonBlock(fds[0]) < 0 || + virSetNonBlock(fds[1]) < 0 || + virSetCloseExec(fds[0]) < 0 || + virSetCloseExec(fds[1]) < 0) { + virReportSystemError(errno, "%s", + _("Failed to setup pipe flags")); + goto error; + } + + if ((srv->sigwatch = virEventAddHandle(fds[0], + VIR_EVENT_HANDLE_READABLE, + virNetServerSignalEvent, + srv, NULL)) < 0) { + virNetError(VIR_ERR_INTERNAL_ERROR, "%s", + _("Failed to add signal handle watch")); + goto error; + } + + srv->sigread = fds[0]; + srv->sigwrite = fds[1]; + sigWrite = fds[1]; + + return 0; + +error: + VIR_FORCE_CLOSE(fds[0]); + VIR_FORCE_CLOSE(fds[1]); + return -1; +} + +int virNetServerAddSignalHandler(virNetServerPtr srv, + int signum, + virNetServerSignalFunc func, + void *opaque) +{ + virNetServerSignalPtr sigdata; + struct sigaction sig_action; + + virNetServerLock(srv); + + if (virNetServerSignalSetup(srv) < 0) + goto error; + + if (VIR_EXPAND_N(srv->signals, srv->nsignals, 1) < 0) + goto no_memory; + + if (VIR_ALLOC(sigdata) < 0) + goto no_memory; + + sigdata->signum = signum; + sigdata->func = func; + sigdata->opaque = opaque; + + memset(&sig_action, 0, sizeof(sig_action)); + sig_action.sa_sigaction = virNetServerSignalHandler; + sig_action.sa_flags = SA_SIGINFO; + sigemptyset(&sig_action.sa_mask); + + sigaction(signum, &sig_action, &sigdata->oldaction); + + srv->signals[srv->nsignals-1] = sigdata; + + virNetServerUnlock(srv); + return 0; + +no_memory: + virReportOOMError(); +error: + VIR_FREE(sigdata); + virNetServerUnlock(srv); + return -1; +} + + + +int virNetServerAddService(virNetServerPtr srv, + virNetServerServicePtr svc) +{ + virNetServerLock(srv); + + if (VIR_EXPAND_N(srv->services, srv->nservices, 1) < 0) + goto no_memory; + + srv->services[srv->nservices-1] = svc; + virNetServerServiceRef(svc); + + virNetServerServiceSetDispatcher(svc, + virNetServerDispatchNewClient, + srv); + + virNetServerUnlock(srv); + return 0; + +no_memory: + virReportOOMError(); + virNetServerUnlock(srv); + return -1; +} + +int virNetServerAddProgram(virNetServerPtr srv, + virNetServerProgramPtr prog) +{ + virNetServerLock(srv); + + if (VIR_EXPAND_N(srv->programs, srv->nprograms, 1) < 0) + goto no_memory; + + srv->programs[srv->nprograms-1] = prog; + virNetServerProgramRef(prog); + + virNetServerUnlock(srv); + return 0; + +no_memory: + virReportOOMError(); + virNetServerUnlock(srv); + return -1; +} + +int virNetServerSetTLSContext(virNetServerPtr srv, + virNetTLSContextPtr tls) +{ + srv->tls = tls; + virNetTLSContextRef(tls); + return 0; +} + + +static void virNetServerAutoShutdownTimer(int timerid ATTRIBUTE_UNUSED, + void *opaque) { + virNetServerPtr srv = opaque; + + virNetServerLock(srv); + + if (srv->autoShutdownFunc(srv, srv->autoShutdownOpaque)) { + VIR_DEBUG0("Automatic shutdown triggered"); + srv->quit = 1; + } + + virNetServerUnlock(srv); +} + + +void virNetServerUpdateServices(virNetServerPtr srv, + bool enabled) +{ + int i; + + virNetServerLock(srv); + for (i = 0 ; i < srv->nservices ; i++) + virNetServerServiceToggle(srv->services[i], enabled); + + virNetServerUnlock(srv); +} + + +void virNetServerRun(virNetServerPtr srv) +{ + int timerid = -1; + int timerActive = 0; + + virNetServerLock(srv); + + if (srv->autoShutdownTimeout && + (timerid = virEventAddTimeout(-1, + virNetServerAutoShutdownTimer, + srv, NULL)) < 0) { + virNetError(VIR_ERR_INTERNAL_ERROR, "%s", + _("Failed to register shutdown timeout")); + goto cleanup; + } + + while (!srv->quit) { + /* A shutdown timeout is specified, so check + * if any drivers have active state, if not + * shutdown after timeout seconds + */ + if (srv->autoShutdownTimeout) { + if (timerActive) { + if (srv->clients) { + DEBUG("Deactivating shutdown timer %d", timerid); + virEventUpdateTimeout(timerid, -1); + timerActive = 0; + } + } else { + if (!srv->clients) { + DEBUG("Activating shutdown timer %d", timerid); + virEventUpdateTimeout(timerid, + srv->autoShutdownTimeout * 1000); + timerActive = 1; + } + } + } + + virNetServerUnlock(srv); + if (virEventRunOnce() < 0) { + virNetServerLock(srv); + DEBUG0("Loop iteration error, exiting"); + break; + } + virNetServerLock(srv); + +#if 0 + reprocess: + for (i = 0 ; i < srv->nclients ; i++) { + int inactive; + virMutexLock(&srv->clients[i]->lock); + inactive = srv->clients[i]->fd == -1 + && srv->clients[i]->refs == 0; + virMutexUnlock(&srv->clients[i]->lock); + if (inactive) { + qemudFreeClient(srv->clients[i]); + srv->nclients--; + if (i < srv->nclients) + memmove(srv->clients + i, + srv->clients + i + 1, + sizeof (*srv->clients) * (srv->nclients - i)); + + VIR_SHRINK_N(srv->clients, srv->nclients, 0); + goto reprocess; + } + } +#endif + + } + +cleanup: + virNetServerUnlock(srv); +} + + +void virNetServerQuit(virNetServerPtr srv) +{ + virNetServerLock(srv); + + srv->quit = 1; + + virNetServerUnlock(srv); +} + +void virNetServerFree(virNetServerPtr srv) +{ + int i; + + if (!srv) + return; + + virNetServerLock(srv); + srv->refs--; + if (srv->refs > 0) { + virNetServerUnlock(srv); + return; + } + + for (i = 0 ; i < srv->nservices ; i++) + virNetServerServiceToggle(srv->services[i], false); + + virThreadPoolFree(srv->workers); + + for (i = 0 ; i < srv->nsignals ; i++) { + sigaction(srv->signals[i]->signum, &srv->signals[i]->oldaction, NULL); + VIR_FREE(srv->signals[i]); + } + VIR_FREE(srv->signals); + VIR_FORCE_CLOSE(srv->sigread); + VIR_FORCE_CLOSE(srv->sigwrite); + if (srv->sigwatch > 0) + virEventRemoveHandle(srv->sigwatch); + + for (i = 0 ; i < srv->nservices ; i++) + virNetServerServiceFree(srv->services[i]); + VIR_FREE(srv->services); + + for (i = 0 ; i < srv->nprograms ; i++) + virNetServerProgramFree(srv->programs[i]); + VIR_FREE(srv->programs); + + for (i = 0 ; i < srv->nclients ; i++) + virNetServerClientFree(srv->clients[i]); + VIR_FREE(srv->clients); + + virNetServerUnlock(srv); + virMutexDestroy(&srv->lock); + VIR_FREE(srv); +} + diff --git a/src/rpc/virnetserver.h b/src/rpc/virnetserver.h new file mode 100644 index 0000000..df9e714 --- /dev/null +++ b/src/rpc/virnetserver.h @@ -0,0 +1,74 @@ +/* + * virnetserver.h: generic network RPC server + * + * Copyright (C) 2006-2010 Red Hat, Inc. + * Copyright (C) 2006 Daniel P. Berrange + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA + * + * Author: Daniel P. Berrange <berrange@redhat.com> + */ + +#ifndef __VIR_NET_SERVER_H__ +#define __VIR_NET_SERVER_H__ + +#include <stdbool.h> +#include <signal.h> + +#include "virnettlscontext.h" +#include "virnetserverprogram.h" +#include "virnetserverclient.h" +#include "virnetserverservice.h" + +virNetServerPtr virNetServerNew(size_t min_workers, + size_t max_workers, + size_t max_clients); + +typedef int (*virNetServerAutoShutdownFunc)(virNetServerPtr srv, void *opaque); + +void virNetServerRef(virNetServerPtr srv); + +void virNetServerAutoShutdown(virNetServerPtr srv, + unsigned int timeout, + virNetServerAutoShutdownFunc func, + void *opaque); + +typedef void (*virNetServerSignalFunc)(virNetServerPtr srv, siginfo_t *info, void *opaque); + +int virNetServerAddSignalHandler(virNetServerPtr srv, + int signum, + virNetServerSignalFunc func, + void *opaque); + +int virNetServerAddService(virNetServerPtr srv, + virNetServerServicePtr svc); + +int virNetServerAddProgram(virNetServerPtr srv, + virNetServerProgramPtr prog); + +int virNetServerSetTLSContext(virNetServerPtr srv, + virNetTLSContextPtr tls); + +void virNetServerUpdateServices(virNetServerPtr srv, + bool enabled); + +void virNetServerRun(virNetServerPtr srv); + +void virNetServerQuit(virNetServerPtr srv); + +void virNetServerFree(virNetServerPtr srv); + + +#endif diff --git a/src/rpc/virnetserverclient.c b/src/rpc/virnetserverclient.c new file mode 100644 index 0000000..76d4b33 --- /dev/null +++ b/src/rpc/virnetserverclient.c @@ -0,0 +1,974 @@ + +#include <config.h> + +# if HAVE_SASL +# include <sasl/sasl.h> +# endif + +#include "virnetserverclient.h" + +#include "logging.h" +#include "virterror_internal.h" +#include "memory.h" +#include "threads.h" + +#define VIR_FROM_THIS VIR_FROM_RPC + +#define virNetError(code, ...) \ + virReportErrorHelper(NULL, VIR_FROM_RPC, code, __FILE__, \ + __FUNCTION__, __LINE__, __VA_ARGS__) + +/* Allow for filtering of incoming messages to a custom + * dispatch processing queue, instead of client->dx. + */ + +typedef struct _virNetServerClientFilter virNetServerClientFilter; +typedef virNetServerClientFilter *virNetServerClientFilterPtr; + +typedef int (*virNetServerClientFilterFunc)(virNetServerClientPtr client, + virNetServerMessagePtr msg, + void *opaque); + +struct _virNetServerClientFilter { + virNetServerClientFilterFunc func; + void *opaque; + + virNetServerClientFilterPtr next; +}; + + +typedef struct _virNetServerClientStream virNetServerClientStream; +typedef virNetServerClientStream *virNetServerClientStreamPtr; + +struct _virNetServerClientStream { + void *opaque; + + int procedure; + int serial; + + unsigned int recvEOF : 1; + unsigned int closed : 1; + + virNetServerClientFilter filter; + + virNetServerMessagePtr rx; + int tx; + + virNetServerClientStreamPtr next; +}; + +#if HAVE_SASL +/* Whether we're passing reads & writes through a sasl SSF */ +enum virNetServerClientSSF { + VIR_NET_CLIENT_SSF_NONE = 0, + VIR_NET_CLIENT_SSF_READ = 1, + VIR_NET_CLIENT_SSF_WRITE = 2, +}; +#endif + +struct _virNetServerClient +{ + int refs; + + virMutex lock; + virNetSocketPtr sock; + int auth; + virNetTLSContextPtr tlsCtxt; + virNetTLSSessionPtr tls; + unsigned int handshake : 1; + unsigned int closing : 1; + +# if HAVE_SASL + sasl_conn_t *saslconn; + int saslSSF; + const char *saslDecoded; + unsigned int saslDecodedLength; + unsigned int saslDecodedOffset; + const char *saslEncoded; + unsigned int saslEncodedLength; + unsigned int saslEncodedOffset; + char *saslUsername; +# endif + + /* Count of messages in the 'tx' queue, + * and the server worker pool queue + * ie RPC calls in progress. Does not count + * async events which are not used for + * throttling calculations */ + size_t nrequests; + size_t nrequests_max; + /* Zero or one messages being received. Zero if + * nrequests >= max_clients and throttling */ + virNetServerMessagePtr rx; + /* Zero or many messages waiting for transmit + * back to client, including async events */ + virNetServerMessagePtr tx; + + /* Filters to capture messages that would otherwise + * end up on the 'dx' queue */ + virNetServerClientFilterPtr filters; + + /* Data streams */ + virNetServerClientStreamPtr streams; + + virNetServerClientDispatchFunc dispatchFunc; + void *dispatchOpaque; +}; + + +static void virNetServerClientDispatchEvent(virNetSocketPtr sock, int events, void *opaque); +static void virNetServerClientFinishMessageLocked(virNetServerClientPtr client, + virNetServerMessagePtr msg); + +static void virNetServerClientLock(virNetServerClientPtr client) +{ + virMutexLock(&client->lock); +} + +static void virNetServerClientUnlock(virNetServerClientPtr client) +{ + virMutexUnlock(&client->lock); +} + +/* + * @client: a locked client object + */ +static int +virNetServerClientCalculateHandleMode(virNetServerClientPtr client) { + int mode = 0; + + if (client->handshake) { + if (virNetTLSSessionHandshakeDirection(client->tls) == 0) + mode |= VIR_EVENT_HANDLE_READABLE; + else + mode |= VIR_EVENT_HANDLE_WRITABLE; + } else { + /* If there is a message on the rx queue then + * we're wanting more input */ + if (client->rx) + mode |= VIR_EVENT_HANDLE_READABLE; + + /* If there are one or more messages to send back to client, + then monitor for writability on socket */ + if (client->tx) + mode |= VIR_EVENT_HANDLE_WRITABLE; + } + + return mode; +} + +/* + * @server: a locked or unlocked server object + * @client: a locked client object + */ +static int virNetServerClientRegisterEvent(virNetServerClientPtr client) +{ + int mode = virNetServerClientCalculateHandleMode(client); + + VIR_DEBUG("Registering client event callback %d", mode); + if (virNetSocketAddIOCallback(client->sock, + mode, + virNetServerClientDispatchEvent, + client) < 0) + return -1; + + return 0; +} + +/* + * @client: a locked client object + */ +static void virNetServerClientUpdateEvent(virNetServerClientPtr client) +{ + int mode; + + if (!client->sock) + return; + + mode = virNetServerClientCalculateHandleMode(client); + + virNetSocketUpdateIOCallback(client->sock, mode); +} + + +static void virNetServerClientMessageQueuePush(virNetServerMessagePtr *queue, + virNetServerMessagePtr msg) +{ + virNetServerMessagePtr tmp = *queue; + + if (tmp) { + while (tmp->next) + tmp = tmp->next; + tmp->next = msg; + } else { + *queue = msg; + } +} + +static virNetServerMessagePtr +virNetServerClientMessageQueueServe(virNetServerMessagePtr *queue) +{ + virNetServerMessagePtr tmp = *queue; + + if (tmp) { + *queue = tmp->next; + tmp->next = NULL; + } + + return tmp; +} + + +static ssize_t virNetServerClientTLSWriteFunc(const char *buf, size_t len, + void *opaque) +{ + virNetServerClientPtr client = opaque; + + return virNetSocketWrite(client->sock, buf, len); +} + +static ssize_t virNetServerClientTLSReadFunc(char *buf, size_t len, + void *opaque) +{ + virNetServerClientPtr client = opaque; + + return virNetSocketRead(client->sock, buf, len); +} + + +/* Check the client's access. */ +static int +virNetServerClientCheckAccess(virNetServerClientPtr client) +{ + virNetServerMessagePtr confirm; + + /* Verify client certificate. */ + if (virNetTLSContextCheckCertificate(client->tlsCtxt, client->tls) < 0) + return -1; + + if (client->tx) { + VIR_INFO0(_("client had unexpected data pending tx after access check")); + return -1; + } + + if (VIR_ALLOC(confirm) < 0) { + virReportOOMError(); + return -1; + } + + /* Checks have succeeded. Write a '\1' byte back to the client to + * indicate this (otherwise the socket is abruptly closed). + * (NB. The '\1' byte is sent in an encrypted record). + */ + confirm->async = 1; + confirm->msg.bufferLength = 1; + confirm->msg.bufferOffset = 0; + confirm->msg.buffer[0] = '\1'; + + client->tx = confirm; + + return 0; +} + + +virNetServerClientPtr virNetServerClientNew(virNetSocketPtr sock, + int auth, + virNetTLSContextPtr tls) +{ + virNetServerClientPtr client; + + VIR_DEBUG("sock=%p auth=%d tls=%p", sock, auth, tls); + + if (VIR_ALLOC(client) < 0) { + virReportOOMError(); + return NULL; + } + + if (virMutexInit(&client->lock) < 0) + goto error; + + client->refs = 1; + client->sock = sock; + client->auth = auth; + client->tlsCtxt = tls; + client->nrequests_max = 10; /* XXX */ + + virNetTLSContextRef(tls); + + /* Prepare one for packet receive */ + if (VIR_ALLOC(client->rx) < 0) + goto error; + client->rx->msg.bufferLength = VIR_NET_MESSAGE_LEN_MAX; + + VIR_DEBUG("client=%p", client); + + return client; + +error: + /* XXX ref counting is better than this */ + client->sock = NULL; /* Caller owns 'sock' upon failure */ + virNetServerClientFree(client); + return NULL; +} + +void virNetServerClientRef(virNetServerClientPtr client) +{ + virNetServerClientLock(client); + client->refs++; + virNetServerClientUnlock(client); +} + + +void virNetServerClientSetDispatcher(virNetServerClientPtr client, + virNetServerClientDispatchFunc func, + void *opaque) +{ + virNetServerClientLock(client); + client->dispatchFunc = func; + client->dispatchOpaque = opaque; + virNetServerClientUnlock(client); +} + + +const char *virNetServerClientAddrString(virNetServerClientPtr client) +{ + return virNetSocketRemoteAddrString(client->sock); +} + + +void virNetServerClientFree(virNetServerClientPtr client) +{ + VIR_DEBUG("client=%p", client); + + if (!client) + return; + + virNetServerClientLock(client); + + client->refs--; + if (client->refs > 0) { + virNetServerClientUnlock(client); + return; + } + + while (client->rx) { + virNetServerMessagePtr msg + = virNetServerClientMessageQueueServe(&client->rx); + VIR_FREE(msg); + } + while (client->tx) { + virNetServerMessagePtr msg + = virNetServerClientMessageQueueServe(&client->tx); + VIR_FREE(msg); + } + + virNetTLSSessionFree(client->tls); + virNetTLSContextFree(client->tlsCtxt); + virNetSocketFree(client->sock); + virNetServerClientUnlock(client); + virMutexDestroy(&client->lock); + VIR_FREE(client); +} + + +/* + * You must hold lock for the client + * + * We don't free stuff here, merely disconnect the client's + * network socket & resources. + * + * Full free of the client is done later in a safe point + * where it can be guaranteed it is no longer in use + */ +static void virNetServerClientCloseLocked(virNetServerClientPtr client) +{ + /* Do now, even though we don't close the socket + * until end, to ensure we don't get invoked + * again due to tls shutdown */ + if (client->sock) + virNetSocketRemoveIOCallback(client->sock); + +#if HAVE_SASL + if (client->saslconn) { + sasl_dispose(&client->saslconn); + client->saslconn = NULL; + } + VIR_FREE(client->saslUsername); +#endif + if (client->tls) { + virNetTLSSessionFree(client->tls); + client->tls = NULL; + } + if (client->sock) { + virNetSocketFree(client->sock); + client->sock = NULL; + } + +} + + +/* Client must be unlocked */ +void virNetServerClientClose(virNetServerClientPtr client) +{ + virNetServerClientLock(client); + virNetServerClientCloseLocked(client); + virNetServerClientUnlock(client); +} + + +int virNetServerClientInit(virNetServerClientPtr client) +{ + virNetServerClientLock(client); + + if (!client->tlsCtxt) { + /* Plain socket, so prepare to read first message */ + if (virNetServerClientRegisterEvent(client) < 0) + goto error; + } else { + int ret; + + if (!(client->tls = virNetTLSSessionNew(client->tlsCtxt, + NULL, + virNetServerClientTLSWriteFunc, + virNetServerClientTLSReadFunc, + client))) + goto error; + + /* Begin the TLS handshake. */ + ret = virNetTLSSessionHandshake(client->tls); + if (ret == 0) { + client->handshake = 0; + /* Unlikely, but ... Next step is to check the certificate. */ + if (virNetServerClientCheckAccess(client) < 0) + goto error; + + /* Handshake & cert check OK, so prepare to read first message */ + if (virNetServerClientRegisterEvent(client) < 0) + goto error; + } else if (ret > 0) { + /* Most likely, need to do more handshake data */ + client->handshake = 1; + + if (virNetServerClientRegisterEvent(client) < 0) + goto error; + } else { +#if 0 + PROBE(CLIENT_TLS_FAIL, "fd=%d", client->fd); +#endif + goto error; + } + } + + virNetServerClientUnlock(client); + return 0; + +error: + virNetServerClientUnlock(client); + return -1; +} + + + +/* + * Read data into buffer using wire decoding (plain or TLS) + * + * Returns: + * -1 on error or EOF + * 0 on EAGAIN + * n number of bytes + */ +static ssize_t virNetServerClientReadBuf(virNetServerClientPtr client, + char *data, ssize_t len) +{ + ssize_t ret; + + if (len < 0) { + virNetError(VIR_ERR_RPC, + _("unexpected negative length request %lld"), + (long long int) len); + virNetServerClientCloseLocked(client); + return -1; + } + + /*virNetServerClientDebug ("virNetServerClientRead: len = %d", len);*/ + + if (client->tls) + ret = virNetTLSSessionRead(client->tls, data, len); + else + ret = virNetSocketRead(client->sock, data, len); + + if (ret == -1 && (errno == EAGAIN || + errno == EINTR)) + return 0; + if (ret <= 0) { + if (ret != 0) + virReportSystemError(errno, "%s", + _("Unable to read from client")); + else + VIR_DEBUG0("EOF from client connection"); + virNetServerClientCloseLocked(client); + return -1; + } + + return ret; +} + +/* + * Read data into buffer without decoding + * + * Returns: + * -1 on error or EOF + * 0 on EAGAIN + * n number of bytes + */ +static ssize_t virNetServerClientReadPlain(virNetServerClientPtr client) +{ + ssize_t ret; + ret = virNetServerClientReadBuf(client, + client->rx->msg.buffer + client->rx->msg.bufferOffset, + client->rx->msg.bufferLength - client->rx->msg.bufferOffset); + if (ret <= 0) + return ret; /* -1 error, 0 eagain */ + + client->rx->msg.bufferOffset += ret; + return ret; +} + +#if HAVE_SASL +/* + * Read data into buffer decoding with SASL + * + * Returns: + * -1 on error or EOF + * 0 on EAGAIN + * n number of bytes + */ +static ssize_t virNetServerClientReadSASL(virNetServerClientPtr client) +{ + ssize_t got, want; + + /* We're doing a SSF data read, so now its times to ensure + * future writes are under SSF too. + * + * cf remoteSASLCheckSSF in remote.c + */ + client->saslSSF |= VIR_NET_CLIENT_SSF_WRITE; + + /* Need to read some more data off the wire */ + if (client->saslDecoded == NULL) { + int ret; + char encoded[8192]; + ssize_t encodedLen = sizeof(encoded); + encodedLen = virNetServerClientReadBuf(client, encoded, encodedLen); + + if (encodedLen <= 0) + return encodedLen; + + ret = sasl_decode(client->saslconn, encoded, encodedLen, + &client->saslDecoded, &client->saslDecodedLength); + if (ret != SASL_OK) { + virNetError(VIR_ERR_INTERNAL_ERROR, + _("failed to decode SASL data %s"), + sasl_errstring(ret, NULL, NULL)); + virNetServerClientCloseLocked(client); + return -1; + } + + client->saslDecodedOffset = 0; + } + + /* Some buffered decoded data to return now */ + got = client->saslDecodedLength - client->saslDecodedOffset; + want = client->rx->msg.bufferLength - client->rx->msg.bufferOffset; + + if (want > got) + want = got; + + memcpy(client->rx->msg.buffer + client->rx->msg.bufferOffset, + client->saslDecoded + client->saslDecodedOffset, want); + client->saslDecodedOffset += want; + client->rx->msg.bufferOffset += want; + + if (client->saslDecodedOffset == client->saslDecodedLength) { + client->saslDecoded = NULL; + client->saslDecodedOffset = client->saslDecodedLength = 0; + } + + return want; +} +#endif + +/* + * Read as much data off wire as possible till we fill our + * buffer, or would block on I/O + */ +static ssize_t virNetServerClientRead(virNetServerClientPtr client) +{ +#if HAVE_SASL + if (client->saslSSF & VIR_NET_CLIENT_SSF_READ) + return virNetServerClientReadSASL(client); + else +#endif + return virNetServerClientReadPlain(client); +} + + +/* + * Read data until we get a complete message to process + */ +static void virNetServerClientDispatchRead(virNetServerClientPtr client) +{ +readmore: + if (virNetServerClientRead(client) < 0) + return; /* Error */ + + if (client->rx->msg.bufferOffset < client->rx->msg.bufferLength) + return; /* Still not read enough */ + + /* Either done with length word header */ + if (client->rx->msg.bufferLength == VIR_NET_MESSAGE_LEN_MAX) { + if (virNetMessageDecodeLength(&client->rx->msg) < 0) + return; + + virNetServerClientUpdateEvent(client); + + /* Try and read payload immediately instead of going back + into poll() because chances are the data is already + waiting for us */ + goto readmore; + } else { + /* Grab the completed message */ + virNetServerMessagePtr msg = virNetServerClientMessageQueueServe(&client->rx); + virNetServerClientFilterPtr filter; + + /* Decode the header so we can use it for routing decisions */ + if (virNetMessageDecodeHeader(&msg->msg) < 0) { + VIR_FREE(msg); + virNetServerClientCloseLocked(client); + } + + /* Check if any filters match this message */ + filter = client->filters; + while (filter) { + int ret; + ret = (filter->func)(client, msg, filter->opaque); + if (ret == 1) { + msg = NULL; + break; + } else if (ret == -1) { + VIR_FREE(msg); + virNetServerClientCloseLocked(client); + return; + } + filter = filter->next; + } + + client->nrequests++; + + /* Possibly need to create another receive buffer */ + if ((client->nrequests < client->nrequests_max) && + VIR_ALLOC(client->rx) < 0) { + virNetServerClientCloseLocked(client); + } else { + if (client->rx) + client->rx->msg.bufferLength = VIR_NET_MESSAGE_LEN_MAX; + + virNetServerClientUpdateEvent(client); + + } + + /* Send it off for processing */ + if (msg) { + if (client->dispatchFunc) + client->dispatchFunc(client, msg, client->dispatchOpaque); + else + virNetServerClientFinishMessageLocked(client, msg); + } + } +} + + +/* + * Send a chunk of data using wire encoding (plain or TLS) + * + * Returns: + * -1 on error + * 0 on EAGAIN + * n number of bytes + */ +static ssize_t virNetServerClientWriteBuf(virNetServerClientPtr client, + const char *data, ssize_t len) +{ + ssize_t ret; + + if (len < 0) { + virNetError(VIR_ERR_INTERNAL_ERROR, + _("unexpected negative length request %lld"), + (long long int) len); + virNetServerClientCloseLocked(client); + return -1; + } + + if (client->tls) + ret = virNetTLSSessionWrite(client->tls, data, len); + else + ret = virNetSocketWrite(client->sock, data, len); + + if (ret == -1 && (errno == EAGAIN || + errno == EINTR)) + return 0; + if (ret == -1) { + virReportSystemError(errno, "%s", + _("Unable to write to client")); + virNetServerClientCloseLocked(client); + return -1; + } + return ret; +} + + +/* + * Send client->tx using no encoding + * + * Returns: + * -1 on error or EOF + * 0 on EAGAIN + * n number of bytes + */ +static int virNetServerClientWritePlain(virNetServerClientPtr client) +{ + int ret = virNetServerClientWriteBuf(client, + client->tx->msg.buffer + client->tx->msg.bufferOffset, + client->tx->msg.bufferLength - client->tx->msg.bufferOffset); + if (ret <= 0) + return ret; /* -1 error, 0 = egain */ + client->tx->msg.bufferOffset += ret; + return ret; +} + + +#if HAVE_SASL +/* + * Send client->tx using SASL encoding + * + * Returns: + * -1 on error + * 0 on EAGAIN + * n number of bytes + */ +static int virNetServerClientWriteSASL(virNetServerClientPtr client) +{ + int ret; + + /* Not got any pending encoded data, so we need to encode raw stuff */ + if (client->saslEncoded == NULL) { + ret = sasl_encode(client->saslconn, + client->tx->msg.buffer + client->tx->msg.bufferOffset, + client->tx->msg.bufferLength - client->tx->msg.bufferOffset, + &client->saslEncoded, + &client->saslEncodedLength); + + if (ret != SASL_OK) { + virNetError(VIR_ERR_INTERNAL_ERROR, + _("failed to encode SASL data %s"), + sasl_errstring(ret, NULL, NULL)); + virNetServerClientCloseLocked(client); + return -1; + } + + client->saslEncodedOffset = 0; + } + + /* Send some of the encoded stuff out on the wire */ + ret = virNetServerClientWriteBuf(client, + client->saslEncoded + client->saslEncodedOffset, + client->saslEncodedLength - client->saslEncodedOffset); + + if (ret <= 0) + return ret; /* -1 error, 0 == egain */ + + /* Note how much we sent */ + client->saslEncodedOffset += ret; + + /* Sent all encoded, so update raw buffer to indicate completion */ + if (client->saslEncodedOffset == client->saslEncodedLength) { + client->saslEncoded = NULL; + client->saslEncodedOffset = client->saslEncodedLength = 0; + + /* Mark as complete, so caller detects completion */ + client->tx->msg.bufferOffset = client->tx->msg.bufferLength; + } + + return ret; +} +#endif + +/* + * Send as much data in the client->tx as possible + * + * Returns: + * -1 on error or EOF + * 0 on EAGAIN + * n number of bytes + */ +static ssize_t virNetServerClientWrite(virNetServerClientPtr client) +{ +#if HAVE_SASL + if (client->saslSSF & VIR_NET_CLIENT_SSF_WRITE) + return virNetServerClientWriteSASL(client); + else +#endif + return virNetServerClientWritePlain(client); +} + + +/* + * Process all queued client->tx messages until + * we would block on I/O + */ +static void +virNetServerClientDispatchWrite(virNetServerClientPtr client) +{ + while (client->tx) { + ssize_t ret; + + ret = virNetServerClientWrite(client); + if (ret < 0) { + virNetServerClientCloseLocked(client); + return; + } + if (ret == 0) + return; /* Would block on write EAGAIN */ + + if (client->tx->msg.bufferOffset == client->tx->msg.bufferLength) { + virNetServerMessagePtr reply; + + /* Get finished reply from head of tx queue */ + reply = virNetServerClientMessageQueueServe(&client->tx); + + virNetServerClientFinishMessageLocked(client, reply); + + if (client->closing) + virNetServerClientCloseLocked(client); + } + } +} + +static void +virNetServerClientDispatchHandshake(virNetServerClientPtr client) +{ + int ret; + /* Continue the handshake. */ + ret = virNetTLSSessionHandshake(client->tls); + if (ret == 0) { + client->handshake = 0; + + /* Finished. Next step is to check the certificate. */ + if (virNetServerClientCheckAccess(client) < 0) + virNetServerClientCloseLocked(client); + else + virNetServerClientUpdateEvent(client); + } else if (ret > 0) { + /* Carry on waiting for more handshake. Update + the events just in case handshake data flow + direction has changed */ + virNetServerClientUpdateEvent (client); + } else { +#if 0 + PROBE(CLIENT_TLS_FAIL, "fd=%d", client->fd); +#endif + /* Fatal error in handshake */ + virNetServerClientCloseLocked(client); + } +} + +static void +virNetServerClientDispatchEvent(virNetSocketPtr sock, int events, void *opaque) +{ + virNetServerClientPtr client = opaque; + + virNetServerClientLock(client); + + if (client->sock != sock) { + virNetSocketRemoveIOCallback(sock); + virNetServerClientUnlock(client); + return; + } + + if (events & (VIR_EVENT_HANDLE_WRITABLE | + VIR_EVENT_HANDLE_READABLE)) { + if (client->handshake) { + virNetServerClientDispatchHandshake(client); + } else { + if (events & VIR_EVENT_HANDLE_WRITABLE) + virNetServerClientDispatchWrite(client); + if (events & VIR_EVENT_HANDLE_READABLE) + virNetServerClientDispatchRead(client); + } + } + + /* NB, will get HANGUP + READABLE at same time upon + * disconnect */ + if (events & (VIR_EVENT_HANDLE_ERROR | + VIR_EVENT_HANDLE_HANGUP)) + virNetServerClientCloseLocked(client); + + virNetServerClientUnlock(client); +} + + +void virNetServerClientSendMessage(virNetServerClientPtr client, + virNetServerMessagePtr msg) +{ + virNetServerClientLock(client); + + virNetServerClientMessageQueuePush(&client->tx, msg); + + virNetServerClientUpdateEvent(client); + + virNetServerClientUnlock(client); +} + +static void virNetServerClientFinishMessageLocked(virNetServerClientPtr client, + virNetServerMessagePtr msg) +{ + if (msg->streamTX) { +#if 0 + XXX + remoteStreamMessageFinished(client, msg); +#endif + } else if (!msg->async) + client->nrequests--; + + /* See if the recv queue is currently throttled */ + if (!client->rx && + client->nrequests < client->nrequests_max) { + /* Reset message record for next RX attempt */ + memset(msg, 0, sizeof(*msg)); + client->rx = msg; + /* Get ready to receive next message */ + client->rx->msg.bufferLength = VIR_NET_MESSAGE_LEN_MAX; + } else { + VIR_FREE(msg); + } + + virNetServerClientUpdateEvent(client); +} + +void virNetServerClientFinishMessage(virNetServerClientPtr client, + virNetServerMessagePtr msg) +{ + virNetServerClientLock(client); + virNetServerClientFinishMessageLocked(client, msg); + virNetServerClientUnlock(client); +} + +bool virNetServerClientNeedAuth(virNetServerClientPtr client) +{ + bool need = false; + virNetServerClientLock(client); + if (client->auth) + need = true; + virNetServerClientUnlock(client); + return need; +} diff --git a/src/rpc/virnetserverclient.h b/src/rpc/virnetserverclient.h new file mode 100644 index 0000000..5fe239c --- /dev/null +++ b/src/rpc/virnetserverclient.h @@ -0,0 +1,40 @@ + + +#ifndef __VIR_NET_SERVER_CLIENT_H__ +#define __VIR_NET_SERVER_CLIENT_H__ + +#include "virnetsocket.h" +#include "virnetserverprogram.h" +#include "virnettlscontext.h" + +typedef int (*virNetServerClientDispatchFunc)(virNetServerClientPtr client, + virNetServerMessagePtr msg, + void *opaque); + +virNetServerClientPtr virNetServerClientNew(virNetSocketPtr sock, + int auth, + virNetTLSContextPtr tls); + +void virNetServerClientRef(virNetServerClientPtr client); + +void virNetServerClientSetDispatcher(virNetServerClientPtr client, + virNetServerClientDispatchFunc func, + void *opaque); +void virNetServerClientClose(virNetServerClientPtr client); + +int virNetServerClientInit(virNetServerClientPtr client); + +const char *virNetServerClientAddrString(virNetServerClientPtr client); + +void virNetServerClientSendMessage(virNetServerClientPtr client, + virNetServerMessagePtr msg); + +void virNetServerClientFinishMessage(virNetServerClientPtr client, + virNetServerMessagePtr msg); + +bool virNetServerClientNeedAuth(virNetServerClientPtr client); + +void virNetServerClientFree(virNetServerClientPtr client); + + +#endif /* __VIR_NET_SERVER_CLIENT_H__ */ diff --git a/src/rpc/virnetservermessage.h b/src/rpc/virnetservermessage.h new file mode 100644 index 0000000..88624fe --- /dev/null +++ b/src/rpc/virnetservermessage.h @@ -0,0 +1,20 @@ + +#ifndef __VIR_NET_SERVER_MESSAGE_H__ +#define __VIR_NET_SERVER_MESSAGE_H__ + +#include "virnetmessage.h" + +typedef struct _virNetServerMessage virNetServerMessage; +typedef virNetServerMessage *virNetServerMessagePtr; + +struct _virNetServerMessage { + virNetMessage msg; + + unsigned int async : 1; + unsigned int streamTX : 1; + + virNetServerMessagePtr next; +}; + +#endif /* __VIR_NET_SERVER_MESSAGE_H__ */ + diff --git a/src/rpc/virnetserverprogram.c b/src/rpc/virnetserverprogram.c new file mode 100644 index 0000000..59328ac --- /dev/null +++ b/src/rpc/virnetserverprogram.c @@ -0,0 +1,437 @@ + + +#include <config.h> + +#include "virnetserverprogram.h" +#include "virnetserverclient.h" + +#include "memory.h" +#include "virterror_internal.h" +#include "logging.h" + +#define VIR_FROM_THIS VIR_FROM_RPC + +struct _virNetServerProgram { + int refs; + + unsigned program; + unsigned version; + virNetServerProgramProcPtr procs; + size_t nprocs; + virNetServerProgramErrorHanderPtr err; +}; + +virNetServerProgramPtr virNetServerProgramNew(unsigned program, + unsigned version, + virNetServerProgramProcPtr procs, + size_t nprocs, + virNetServerProgramErrorHanderPtr err) +{ + virNetServerProgramPtr prog; + + if (VIR_ALLOC(prog) < 0) { + virReportOOMError(); + return NULL; + } + + prog->refs = 1; + prog->program = program; + prog->version = version; + prog->procs = procs; + prog->nprocs = nprocs; + prog->err = err; + + return prog; +} + + +void virNetServerProgramRef(virNetServerProgramPtr prog) +{ + prog->refs++; +} + + +int virNetServerProgramMatches(virNetServerProgramPtr prog, + virNetServerMessagePtr msg) +{ + if (prog->program == msg->msg.header.prog && + prog->version == msg->msg.header.vers) + return 1; + return 0; +} + + +static virNetServerProgramProcPtr virNetServerProgramGetProc(virNetServerProgramPtr prog, + int procedure) +{ + if (procedure < 0) + return NULL; + if (procedure >= prog->nprocs) + return NULL; + + return &prog->procs[procedure]; +} + +static void +virNetServerProgramStringError(virNetServerProgramPtr prog, + void *rerr, + int code, + const char *str) +{ + prog->err->func(prog, rerr, code, str); +} + +static void +virNetServerProgramFormatError(virNetServerProgramPtr prog, + void *rerr, + int code, + const char *fmt, + ...) +{ + va_list args; + char msgbuf[1024]; + char *msg = msgbuf; + + va_start(args, fmt); + vsnprintf(msgbuf, sizeof msgbuf, fmt, args); + va_end(args); + + virNetServerProgramStringError(prog, rerr, code, msg); +} + + +#if 0 +static void +virNetServerProgramGenericError(virNetServerProgramPtr prog, + void *rerr) +{ + virNetServerProgramStringError(prog, rerr, + VIR_ERR_INTERNAL_ERROR, + _("function returned error code but did not set an error message")); +} +#endif + + +static void +virNetServerProgramOOMError(virNetServerProgramPtr prog, + void *rerr) +{ + virNetServerProgramStringError(prog, rerr, + VIR_ERR_NO_MEMORY, + _("out of memory")); +} + + +static int +remoteSerializeError(virNetServerProgramPtr prog, + virNetServerClientPtr client, + void *rerr, + int program, + int version, + int procedure, + int type, + int serial) +{ + virNetServerMessagePtr msg = NULL; + + DEBUG("prog=%d ver=%d proc=%d type=%d serial=%d, msg=%p", + program, version, procedure, type, serial, rerr); + + if (VIR_ALLOC(msg) < 0) { + virReportOOMError(); + return -1; + } + + + /* Return header. */ + msg->msg.header.prog = program; + msg->msg.header.vers = version; + msg->msg.header.proc = procedure; + msg->msg.header.type = type; + msg->msg.header.serial = serial; + msg->msg.header.status = VIR_NET_ERROR; + + if (virNetMessageEncodeHeader(&msg->msg) < 0) + goto error; + + if (virNetMessageEncodePayload(&msg->msg, prog->err->filter, rerr) < 0) + goto error; + + /* Put reply on end of tx queue to send out */ + virNetServerClientSendMessage(client, msg); + xdr_free(prog->err->filter, rerr); + + return 0; + +error: + VIR_WARN("Failed to serialize remote error '%p'", rerr); + VIR_FREE(msg); + xdr_free(prog->err->filter, rerr); + return -1; +} + + +/* + * @client: the client to send the error to + * @rerr: the error object to send + * @req: the message this error is in reply to + * + * Send an error message to the client + * + * Returns 0 if the error was sent, -1 upon fatal error + */ +static int +virNetServerProgramSerializeReplyError(virNetServerProgramPtr prog, + virNetServerClientPtr client, + void *rerr, + virNetMessageHeaderPtr req) +{ + /* + * For data streams, errors are sent back as data streams + * For method calls, errors are sent back as method replies + */ + return remoteSerializeError(prog, + client, + rerr, + req->prog, + req->vers, + req->proc, + req->type == VIR_NET_STREAM ? VIR_NET_STREAM : VIR_NET_REPLY, + req->serial); +} + +static int +virNetServerProgramDispatchCall(virNetServerProgramPtr prog, + void *rerr, + virNetServerPtr server, + virNetServerClientPtr client, + virNetServerMessagePtr msg); + +/* + * @server: the unlocked server object + * @client: the locked client object + * @msg: the complete incoming message packet, with header already decoded + * + * This function gets called from qemud when it pulls a incoming + * remote protocol message off the dispatch queue for processing. + * + * The @msg parameter must have had its header decoded already by + * calling remoteDecodeClientMessageHeader + * + * Returns 0 if the message was dispatched, -1 upon fatal error + */ +int virNetServerProgramDispatch(virNetServerProgramPtr prog, + virNetServerPtr server, + virNetServerClientPtr client, + virNetServerMessagePtr msg) +{ + int ret = -1; + char *rerr; + + if (VIR_ALLOC_N(rerr, prog->err->len) < 0) { + virReportOOMError(); + return -1; + } + + DEBUG("prog=%d ver=%d type=%d status=%d serial=%d proc=%d", + msg->msg.header.prog, msg->msg.header.vers, msg->msg.header.type, + msg->msg.header.status, msg->msg.header.serial, msg->msg.header.proc); + + /* Check version, etc. */ + if (msg->msg.header.prog != prog->program) { + virNetServerProgramFormatError(prog, rerr, VIR_ERR_RPC, + _("program mismatch (actual %x, expected %x)"), + msg->msg.header.prog, prog->program); + goto error; + } + + if (msg->msg.header.vers != prog->version) { + virNetServerProgramFormatError(prog, rerr, VIR_ERR_RPC, + _("version mismatch (actual %x, expected %x)"), + msg->msg.header.vers, prog->version); + goto error; + } + + switch (msg->msg.header.type) { + case VIR_NET_CALL: + ret = virNetServerProgramDispatchCall(prog, rerr, server, client, msg); + break; + + case VIR_NET_STREAM: + /* Since stream data is non-acked, async, we may continue to received + * stream packets after we closed down a stream. Just drop & ignore + * these. + */ + VIR_INFO("Ignoring unexpected stream data serial=%d proc=%d status=%d", + msg->msg.header.serial, msg->msg.header.proc, msg->msg.header.status); + virNetServerClientFinishMessage(client, msg); + ret = 0; + break; + + default: + virNetServerProgramFormatError(prog, rerr, VIR_ERR_RPC, + _("Unexpected message type %d"), + (int)msg->msg.header.type); + goto error; + } + + VIR_FREE(rerr); + + return ret; + +error: + ret = virNetServerProgramSerializeReplyError(prog, client, rerr, &msg->msg.header); + + if (ret >= 0) + VIR_FREE(msg); + + VIR_FREE(rerr); + + return ret; +} + + +/* + * @server: the unlocked server object + * @client: the unlocked client object + * @msg: the complete incoming method call, with header already decoded + * + * This method is used to dispatch an message representing an + * incoming method call from a client. It decodes the payload + * to obtain method call arguments, invokves the method and + * then sends a reply packet with the return values + * + * Returns 0 if the reply was sent, or -1 upon fatal error + */ +static int +virNetServerProgramDispatchCall(virNetServerProgramPtr prog, + void *rerr, + virNetServerPtr server, + virNetServerClientPtr client, + virNetServerMessagePtr msg) +{ + char *arg = NULL; + char *ret = NULL; + int rv = -1; + unsigned int len; + virNetServerProgramProcPtr dispatcher; + + memset(rerr, 0, sizeof rerr); + + if (msg->msg.header.status != VIR_NET_OK) { + virNetServerProgramFormatError(prog, rerr, VIR_ERR_RPC, + _("Unexpected message status %d"), + (int)msg->msg.header.status); + goto error; + } + + dispatcher = virNetServerProgramGetProc(prog, msg->msg.header.proc); + + if (!dispatcher) { + virNetServerProgramFormatError(prog, rerr, VIR_ERR_RPC, + _("unknown procedure: %d"), + msg->msg.header.proc); + goto error; + } + + /* If client is marked as needing auth, don't allow any RPC ops + * which are except for authentication ones + */ + if (virNetServerClientNeedAuth(client) && + dispatcher->needAuth) { + /* Explicitly *NOT* calling remoteDispatchAuthError() because + we want back-compatability with libvirt clients which don't + support the VIR_ERR_AUTH_FAILED error code */ + virNetServerProgramFormatError(prog, rerr, VIR_ERR_RPC, + "%s", _("authentication required")); + goto error; + } + + if (VIR_ALLOC_N(arg, dispatcher->arg_len) < 0) { + virNetServerProgramOOMError(prog, rerr); + goto error; + } + if (VIR_ALLOC_N(ret, dispatcher->ret_len) < 0) { + virNetServerProgramOOMError(prog, rerr); + goto error; + } + + if (virNetMessageDecodePayload(&msg->msg, dispatcher->arg_filter, arg) < 0) + goto error; + + /* + * When the RPC handler is called: + * + * - Server object is unlocked + * - Client object is unlocked + * + * Without locking, it is safe to use: + * + * 'rerr', 'args and 'ret' + */ + rv = (dispatcher->func)(server, client, &msg->msg.header, rerr, arg, ret); + + xdr_free(dispatcher->arg_filter, arg); + + if (rv < 0) + goto error; + + /* Return header. We're re-using same message object, so + * only need to tweak type/status fields */ + /*msg->msg.header.prog = msg->msg.header.prog;*/ + /*msg->msg.header.vers = msg->msg.header.vers;*/ + /*msg->msg.header.proc = msg->msg.header.proc;*/ + msg->msg.header.type = VIR_NET_REPLY; + /*msg->msg.header.serial = msg->msg.header.serial;*/ + msg->msg.header.status = VIR_NET_OK; + + if (virNetMessageEncodeHeader(&msg->msg) < 0) { + xdr_free(dispatcher->ret_filter, ret); + goto error; + } + + if (virNetMessageEncodePayload(&msg->msg, dispatcher->ret_filter, ret) < 0) { + xdr_free(dispatcher->ret_filter, ret); + goto error; + } + + /* Reset ready for I/O */ + msg->msg.bufferLength = len; + msg->msg.bufferOffset = 0; + + VIR_FREE(arg); + VIR_FREE(ret); + + /* Put reply on end of tx queue to send out */ + virNetServerClientSendMessage(client, msg); + + return 0; + +error: + /* Bad stuff (de-)serializing message, but we have an + * RPC error message we can send back to the client */ + rv = virNetServerProgramSerializeReplyError(prog, client, rerr, &msg->msg.header); + + if (rv >= 0) + VIR_FREE(msg); + + VIR_FREE(arg); + VIR_FREE(ret); + + return rv; +} + + +void virNetServerProgramFree(virNetServerProgramPtr prog) +{ + if (!prog) + return; + + prog->refs--; + if (prog->refs > 0) + return; + + VIR_FREE(prog); +} + + diff --git a/src/rpc/virnetserverprogram.h b/src/rpc/virnetserverprogram.h new file mode 100644 index 0000000..d116867 --- /dev/null +++ b/src/rpc/virnetserverprogram.h @@ -0,0 +1,76 @@ + +#ifndef __VIR_NET_PROGRAM_H__ +#define __VIR_NET_PROGRAM_H__ + +#include <stdbool.h> + +#include "virnetservermessage.h" + +typedef struct _virNetServer virNetServer; +typedef virNetServer *virNetServerPtr; + +typedef struct _virNetServerClient virNetServerClient; +typedef virNetServerClient *virNetServerClientPtr; + +typedef struct _virNetServerService virNetServerService; +typedef virNetServerService *virNetServerServicePtr; + +typedef struct _virNetServerProgram virNetServerProgram; +typedef virNetServerProgram *virNetServerProgramPtr; + +typedef struct _virNetServerProgramProc virNetServerProgramProc; +typedef virNetServerProgramProc *virNetServerProgramProcPtr; + +typedef struct _virNetServerProgramErrorHandler virNetServerProgramErrorHander; +typedef virNetServerProgramErrorHander *virNetServerProgramErrorHanderPtr; + +typedef int (*virNetServerProgramErrorFunc)(virNetServerProgramPtr prog, + void *rerr, + int code, + const char *msg); + +struct _virNetServerProgramErrorHandler { + virNetServerProgramErrorFunc func; + size_t len; + xdrproc_t filter; +}; + + +typedef int (*virNetServerProgramDispatchFunc)(virNetServerPtr server, + virNetServerClientPtr client, + virNetMessageHeader *hdr, + void *err, + void *args, + void *ret); + +struct _virNetServerProgramProc { + virNetServerProgramDispatchFunc func; + size_t arg_len; + xdrproc_t arg_filter; + size_t ret_len; + xdrproc_t ret_filter; + bool needAuth; +}; + +virNetServerProgramPtr virNetServerProgramNew(unsigned program, + unsigned version, + virNetServerProgramProcPtr procs, + size_t nprocs, + virNetServerProgramErrorHanderPtr err); + +void virNetServerProgramRef(virNetServerProgramPtr prog); + +int virNetServerProgramMatches(virNetServerProgramPtr prog, + virNetServerMessagePtr msg); + +int virNetServerProgramDispatch(virNetServerProgramPtr prog, + virNetServerPtr server, + virNetServerClientPtr client, + virNetServerMessagePtr msg); + +void virNetServerProgramFree(virNetServerProgramPtr prog); + + + + +#endif /* __VIR_NET_SERVER_PROGRAM_H__ */ diff --git a/src/rpc/virnetserverservice.c b/src/rpc/virnetserverservice.c new file mode 100644 index 0000000..f985603 --- /dev/null +++ b/src/rpc/virnetserverservice.c @@ -0,0 +1,208 @@ + +#include <config.h> + + +#include "virnetserverservice.h" + +#include "memory.h" +#include "virterror_internal.h" + + +#define VIR_FROM_THIS VIR_FROM_RPC + +struct _virNetServerService { + int refs; + + size_t nsocks; + virNetSocketPtr *socks; + + int auth; + + virNetTLSContextPtr tls; + + virNetServerServiceDispatchFunc dispatchFunc; + void *dispatchOpaque; +}; + + + +static void virNetServerServiceAccept(virNetSocketPtr sock, + int events ATTRIBUTE_UNUSED, + void *opaque) +{ + virNetServerServicePtr svc = opaque; + virNetServerClientPtr client = NULL; + virNetSocketPtr clientsock = NULL; + + if (virNetSocketAccept(sock, &clientsock) < 0) + goto error; + + if (!clientsock) /* Connection already went away */ + goto cleanup; + + if (!(client = virNetServerClientNew(clientsock, + svc->auth, + svc->tls))) + goto error; + + if (!svc->dispatchFunc) + goto error; + + svc->dispatchFunc(svc, client, svc->dispatchOpaque); + + virNetServerClientFree(client); + +cleanup: + return; + +error: + virNetSocketFree(clientsock); +} + + +virNetServerServicePtr virNetServerServiceNewTCP(const char *nodename, + const char *service, + int auth, + virNetTLSContextPtr tls) +{ + virNetServerServicePtr svc; + int i; + + if (VIR_ALLOC(svc) < 0) + goto no_memory; + + svc->refs = 1; + svc->auth = auth; + svc->tls = tls; + if (tls) + virNetTLSContextRef(tls); + + if (virNetSocketNewListenTCP(nodename, + service, + &svc->socks, + &svc->nsocks) < 0) + goto error; + + for (i = 0 ; i < svc->nsocks ; i++) { + if (virNetSocketListen(svc->socks[i]) < 0) + goto error; + + /* IO callback is initially disabled, until we're ready + * to deal with incoming clients */ + if (virNetSocketAddIOCallback(svc->socks[i], + 0, + virNetServerServiceAccept, + svc) < 0) + goto error; + } + + + return svc; + +no_memory: + virReportOOMError(); +error: + virNetServerServiceFree(svc); + return NULL; +} + + +virNetServerServicePtr virNetServerServiceNewUNIX(const char *path, + mode_t mask, + gid_t grp, + int auth, + virNetTLSContextPtr tls) +{ + virNetServerServicePtr svc; + int i; + + if (VIR_ALLOC(svc) < 0) + goto no_memory; + + svc->refs = 1; + svc->auth = auth; + svc->tls = tls; + if (tls) + virNetTLSContextRef(tls); + + svc->nsocks = 1; + if (VIR_ALLOC_N(svc->socks, svc->nsocks) < 0) + goto no_memory; + + if (virNetSocketNewListenUNIX(path, + mask, + grp, + &svc->socks[0]) < 0) + goto error; + + for (i = 0 ; i < svc->nsocks ; i++) { + if (virNetSocketListen(svc->socks[i]) < 0) + goto error; + + /* IO callback is initially disabled, until we're ready + * to deal with incoming clients */ + if (virNetSocketAddIOCallback(svc->socks[i], + 0, + virNetServerServiceAccept, + svc) < 0) + goto error; + } + + + return svc; + +no_memory: + virReportOOMError(); +error: + virNetServerServiceFree(svc); + return NULL; +} + + +void virNetServerServiceRef(virNetServerServicePtr svc) +{ + svc->refs++; +} + + +void virNetServerServiceSetDispatcher(virNetServerServicePtr svc, + virNetServerServiceDispatchFunc func, + void *opaque) +{ + svc->dispatchFunc = func; + svc->dispatchOpaque = opaque; +} + + +void virNetServerServiceFree(virNetServerServicePtr svc) +{ + int i; + + if (!svc) + return; + + svc->refs--; + if (svc->refs > 0) + return; + + for (i = 0 ; i < svc->nsocks ; i++) + virNetSocketFree(svc->socks[i]); + VIR_FREE(svc->socks); + + virNetTLSContextFree(svc->tls); + + VIR_FREE(svc); +} + +void virNetServerServiceToggle(virNetServerServicePtr svc, + bool enabled) +{ + int i; + + for (i = 0 ; i < svc->nsocks ; i++) + virNetSocketUpdateIOCallback(svc->socks[i], + enabled ? + VIR_EVENT_HANDLE_READABLE : + 0); +} + diff --git a/src/rpc/virnetserverservice.h b/src/rpc/virnetserverservice.h new file mode 100644 index 0000000..4bedde7 --- /dev/null +++ b/src/rpc/virnetserverservice.h @@ -0,0 +1,32 @@ + +#ifndef __VIR_NET_SERVER_SERVICE_H__ +#define __VIR_NET_SERVER_SERVICE_H__ + +#include "virnetserverclient.h" + +typedef int (*virNetServerServiceDispatchFunc)(virNetServerServicePtr svc, + virNetServerClientPtr client, + void *opaque); + +virNetServerServicePtr virNetServerServiceNewTCP(const char *nodename, + const char *service, + int auth, + virNetTLSContextPtr tls); +virNetServerServicePtr virNetServerServiceNewUNIX(const char *path, + mode_t mask, + gid_t grp, + int auth, + virNetTLSContextPtr tls); + +void virNetServerServiceRef(virNetServerServicePtr svc); + +void virNetServerServiceSetDispatcher(virNetServerServicePtr svc, + virNetServerServiceDispatchFunc func, + void *opaque); + +void virNetServerServiceFree(virNetServerServicePtr svc); + +void virNetServerServiceToggle(virNetServerServicePtr svc, + bool enabled); + +#endif -- 1.7.2.3

To facilitate creation of new clients using XDR RPC services, pull alot of the remote driver code into a set of reusable objects. - virNetClient: Encapsulates a socket connection to a remote RPC server. Handles all the network I/O for reading/writing RPC messages. Delegates RPC encoding and decoding to the registered programs - virNetClientProgram: Handles processing and dispatch of RPC messages for a single RPC (program,version). A program can register to receive async events from a client - virNetClientSASLContext: Handles everything todo with SASL authentication and encryption. The callers no longer need directly call any cyrus-sasl APIs, which means error handling is simpler & alternative SASL impls can be provided for Win32 Each new client program now merely needs to define the list of RPC procedures & events it wants and their handlers. It does not need to deal with any of the network I/O functionality at all. --- src/Makefile.am | 14 +- src/rpc/virnetclient.c | 1237 +++++++++++++++++++++++++++++++++++++ src/rpc/virnetclient.h | 60 ++ src/rpc/virnetclientprogram.c | 258 ++++++++ src/rpc/virnetclientprogram.h | 71 +++ src/rpc/virnetclientsaslcontext.c | 246 ++++++++ src/rpc/virnetclientsaslcontext.h | 66 ++ 7 files changed, 1951 insertions(+), 1 deletions(-) create mode 100644 src/rpc/virnetclient.c create mode 100644 src/rpc/virnetclient.h create mode 100644 src/rpc/virnetclientprogram.c create mode 100644 src/rpc/virnetclientprogram.h create mode 100644 src/rpc/virnetclientsaslcontext.c create mode 100644 src/rpc/virnetclientsaslcontext.h diff --git a/src/Makefile.am b/src/Makefile.am index e78a0af..4c6efa8 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -1117,7 +1117,7 @@ libvirt_qemu_la_LIBADD = libvirt.la $(CYGWIN_EXTRA_LIBADD) EXTRA_DIST += $(LIBVIRT_QEMU_SYMBOL_FILE) -noinst_LTLIBRARIES += libvirt-net-rpc.la libvirt-net-rpc-server.la +noinst_LTLIBRARIES += libvirt-net-rpc.la libvirt-net-rpc-server.la libvirt-net-rpc-client.la libvirt_net_rpc_la_SOURCES = \ ../daemon/event.c \ @@ -1153,6 +1153,18 @@ libvirt_net_server_la_LDFLAGS = \ libvirt_net_server_la_LIBADD = \ $(CYGWIN_EXTRA_LIBADD) +libvirt_net_client_la_SOURCES = \ + rpc/virnetclientsaslcontext.h rpc/virnetclientsaslcontext.c \ + rpc/virnetclientprogram.h rpc/virnetclientprogram.c \ + rpc/virnetclient.h rpc/virnetclient.c +libvirt_net_client_la_CFLAGS = \ + $(AM_CFLAGS) +libvirt_net_client_la_LDFLAGS = \ + $(AM_LDFLAGS) \ + $(CYGWIN_EXTRA_LDFLAGS) \ + $(MINGW_EXTRA_LDFLAGS)l +libvirt_net_client_la_LIBADD = \ + $(CYGWIN_EXTRA_LIBADD) libexec_PROGRAMS = diff --git a/src/rpc/virnetclient.c b/src/rpc/virnetclient.c new file mode 100644 index 0000000..d3a8740 --- /dev/null +++ b/src/rpc/virnetclient.c @@ -0,0 +1,1237 @@ + + +#include <config.h> + +#include <unistd.h> +#include <poll.h> +#include <signal.h> + +#include "virnetclient.h" +#include "virnetsocket.h" +#include "memory.h" +#include "threads.h" +#include "files.h" +#include "logging.h" +#include "util.h" +#include "virterror_internal.h" + +#define VIR_FROM_THIS VIR_FROM_RPC + +#define virNetError(code, ...) \ + virReportErrorHelper(NULL, VIR_FROM_RPC, code, __FILE__, \ + __FUNCTION__, __LINE__, __VA_ARGS__) + +#ifdef WIN32 +# define pipe(fds) _pipe(fds,4096, _O_BINARY) +#endif + +typedef struct _virNetClientCall virNetClientCall; +typedef virNetClientCall *virNetClientCallPtr; + +enum { + VIR_NET_CLIENT_MODE_WAIT_TX, + VIR_NET_CLIENT_MODE_WAIT_RX, + VIR_NET_CLIENT_MODE_COMPLETE, +}; + +struct _virNetClientCall { + int mode; + + virNetMessagePtr msg; + int expectReply; + + virCond cond; + + +/* remote_error err; */ + + virNetClientCallPtr next; +}; + + +struct _virNetClient { + int refs; + + virMutex lock; + + virNetSocketPtr sock; + + virNetTLSSessionPtr tls; + char *hostname; + + virNetClientProgramPtr *programs; + size_t nprograms; + + /* For incoming message packets */ + virNetMessage msg; + +#if HAVE_SASL + virNetClientSaslContextPtr sasl; + + const char *saslDecoded; + size_t saslDecodedLength; + size_t saslDecodedOffset; + + const char *saslEncoded; + size_t saslEncodedLength; + size_t saslEncodedOffset; +#endif + + /* Self-pipe to wakeup threads waiting in poll() */ + int wakeupSendFD; + int wakeupReadFD; + + /* List of threads currently waiting for dispatch */ + virNetClientCallPtr waitDispatch; +}; + + +static void virNetClientLock(virNetClientPtr client) +{ + virMutexLock(&client->lock); +} + + +static void virNetClientUnlock(virNetClientPtr client) +{ + virMutexUnlock(&client->lock); +} + +static ssize_t virNetClientTLSWriteFunc(const char *buf, size_t len, + void *opaque) +{ + virNetClientPtr client = opaque; + + return virNetSocketWrite(client->sock, buf, len); +} + + +static ssize_t virNetClientTLSReadFunc(char *buf, size_t len, + void *opaque) +{ + virNetClientPtr client = opaque; + + return virNetSocketRead(client->sock, buf, len); +} + + +static void virNetClientIncomingEvent(virNetSocketPtr sock, + int events, + void *opaque); + +static virNetClientPtr virNetClientNew(virNetSocketPtr sock, + const char *hostname) +{ + virNetClientPtr client; + int wakeupFD[2] = { -1, -1 }; + + if (pipe(wakeupFD) < 0) { + virReportSystemError(errno, "%s", + _("unable to make pipe")); + goto error; + } + + if (VIR_ALLOC(client) < 0) + goto no_memory; + + client->refs = 1; + + if (virMutexInit(&client->lock) < 0) + goto error; + + client->sock = sock; + client->wakeupReadFD = wakeupFD[0]; + client->wakeupSendFD = wakeupFD[1]; + wakeupFD[0] = wakeupFD[1] = -1; + + if (hostname && + !(client->hostname = strdup(hostname))) + goto no_memory; + + /* Set up a callback to listen on the socket data */ + if (virNetSocketAddIOCallback(client->sock, + VIR_EVENT_HANDLE_READABLE, + virNetClientIncomingEvent, + client) < 0) + VIR_DEBUG0("Failed to add event watch, disabling events"); + + return client; + +no_memory: + virReportOOMError(); +error: + VIR_FORCE_CLOSE(wakeupFD[0]); + VIR_FORCE_CLOSE(wakeupFD[1]); + virNetClientFree(client); + return NULL; +} + + +virNetClientPtr virNetClientNewUNIX(const char *path, + bool spawnDaemon, + const char *binary) +{ + virNetSocketPtr sock; + + if (virNetSocketNewConnectUNIX(path, spawnDaemon, binary, &sock) < 0) + return NULL; + + return virNetClientNew(sock, NULL); +} + + +virNetClientPtr virNetClientNewTCP(const char *nodename, + const char *service) +{ + virNetSocketPtr sock; + + if (virNetSocketNewConnectTCP(nodename, service, &sock) < 0) + return NULL; + + return virNetClientNew(sock, nodename); +} + +virNetClientPtr virNetClientNewSSH(const char *nodename, + const char *service, + const char *binary, + const char *username, + bool noTTY, + const char *netcat, + const char *path) +{ + virNetSocketPtr sock; + + if (virNetSocketNewConnectSSH(nodename, service, binary, username, noTTY, netcat, path, &sock) < 0) + return NULL; + + return virNetClientNew(sock, NULL); +} + +virNetClientPtr virNetClientNewCommand(const char **cmdargv, + const char **cmdenv) +{ + virNetSocketPtr sock; + + if (virNetSocketNewConnectCommand(cmdargv, cmdenv, &sock) < 0) + return NULL; + + return virNetClientNew(sock, NULL); +} + + +void virNetClientRef(virNetClientPtr client) +{ + virNetClientLock(client); + client->refs++; + virNetClientUnlock(client); +} + + +void virNetClientFree(virNetClientPtr client) +{ + int i; + + if (!client) + return; + + virNetClientLock(client); + client->refs--; + if (client->refs > 0) { + virNetClientUnlock(client); + return; + } + + for (i = 0 ; i < client->nprograms ; i++) + virNetClientProgramFree(client->programs[i]); + VIR_FREE(client->programs); + + VIR_FORCE_CLOSE(client->wakeupSendFD); + VIR_FORCE_CLOSE(client->wakeupReadFD); + + VIR_FREE(client->hostname); + + virNetSocketRemoveIOCallback(client->sock); + virNetSocketFree(client->sock); + virNetTLSSessionFree(client->tls); + virNetClientSaslContextFree(client->sasl); + virNetClientUnlock(client); + virMutexDestroy(&client->lock); + + VIR_FREE(client); +} + + +void virNetClientSetSASLContext(virNetClientPtr client, + virNetClientSaslContextPtr ctxt) +{ + virNetClientLock(client); + client->sasl = ctxt; + virNetClientSaslContextRef(ctxt); + virNetClientUnlock(client); +} + + +int virNetClientSetTLSSession(virNetClientPtr client, + virNetTLSContextPtr tls) +{ + int ret; + char buf[1]; + int len; + struct pollfd fds[1]; +#ifdef HAVE_PTHREAD_SIGMASK + sigset_t oldmask, blockedsigs; + + sigemptyset (&blockedsigs); + sigaddset (&blockedsigs, SIGWINCH); + sigaddset (&blockedsigs, SIGCHLD); + sigaddset (&blockedsigs, SIGPIPE); +#endif + + virNetClientLock(client); + + if (!(client->tls = virNetTLSSessionNew(tls, + client->hostname, + virNetClientTLSWriteFunc, + virNetClientTLSReadFunc, + client))) + goto error; + + for (;;) { + ret = virNetTLSSessionHandshake(client->tls); + + if (ret < 0) + goto error; + if (ret == 0) + break; + + fds[0].fd = virNetSocketFD(client->sock); + fds[0].revents = 0; + if (virNetTLSSessionHandshakeDirection(client->tls) == 0) + fds[0].events = POLLIN; + else + fds[0].events = POLLOUT; + + /* Block SIGWINCH from interrupting poll in curses programs, + * then restore the original signal mask again immediately + * after the call (RHBZ#567931). Same for SIGCHLD and SIGPIPE + * at the suggestion of Paolo Bonzini and Daniel Berrange. + */ +#ifdef HAVE_PTHREAD_SIGMASK + ignore_value(pthread_sigmask(SIG_BLOCK, &blockedsigs, &oldmask)); +#endif + + repoll: + ret = poll(fds, ARRAY_CARDINALITY(fds), -1); + if (ret < 0 && errno == EAGAIN) + goto repoll; + +#ifdef HAVE_PTHREAD_SIGMASK + ignore_value(pthread_sigmask(SIG_BLOCK, &oldmask, NULL)); +#endif + } + + ret = virNetTLSContextCheckCertificate(tls, client->tls); + + if (ret < 0) + goto error; + + /* At this point, the server is verifying _our_ certificate, IP address, + * etc. If we make the grade, it will send us a '\1' byte. + */ + + fds[0].fd = virNetSocketFD(client->sock); + fds[0].revents = 0; + fds[0].events = POLLIN; + +#ifdef HAVE_PTHREAD_SIGMASK + /* Block SIGWINCH from interrupting poll in curses programs */ + ignore_value(pthread_sigmask(SIG_BLOCK, &blockedsigs, &oldmask)); +#endif + + repoll2: + ret = poll(fds, ARRAY_CARDINALITY(fds), -1); + if (ret < 0 && errno == EAGAIN) + goto repoll2; + +#ifdef HAVE_PTHREAD_SIGMASK + ignore_value(pthread_sigmask(SIG_BLOCK, &oldmask, NULL)); +#endif + + len = virNetTLSSessionRead(client->tls, buf, 1); + if (len < 0) { + virReportSystemError(errno, "%s", + _("Unable to read TLS confirmation")); + goto error; + } + if (len != 1 || buf[0] != '\1') { + virNetError(VIR_ERR_RPC, "%s", + _("server verification (of our certificate or IP " + "address) failed")); + goto error; + } + + virNetClientUnlock(client); + return 0; + +error: + virNetTLSSessionFree(client->tls); + client->tls = NULL; + virNetClientUnlock(client); + return -1; +} + +bool virNetClientIsEncrypted(virNetClientPtr client) +{ + bool ret; + virNetClientLock(client); + ret = client->tls || client->sasl ? true : false; + virNetClientUnlock(client); + return ret; +} + + +int virNetClientAddProgram(virNetClientPtr client, + virNetClientProgramPtr prog) +{ + virNetClientLock(client); + + if (VIR_EXPAND_N(client->programs, client->nprograms, 1) < 0) + goto no_memory; + + client->programs[client->nprograms-1] = prog; + virNetClientProgramRef(prog); + + virNetClientUnlock(client); + return 0; + +no_memory: + virReportOOMError(); + virNetClientUnlock(client); + return -1; +} + + +const char *virNetClientLocalAddrString(virNetClientPtr client) +{ + return virNetSocketLocalAddrString(client->sock); +} + +const char *virNetClientRemoteAddrString(virNetClientPtr client) +{ + return virNetSocketRemoteAddrString(client->sock); +} + +int virNetClientGetTLSKeySize(virNetClientPtr client) +{ + int ret = 0; + virNetClientLock(client); + if (client->tls) + ret = virNetTLSSessionGetKeySize(client->tls); + virNetClientUnlock(client); + return ret; +} + +static int +virNetClientCallDispatchReply(virNetClientPtr client) +{ + virNetClientCallPtr thecall; + + /* Ok, definitely got an RPC reply now find + out who's been waiting for it */ + thecall = client->waitDispatch; + while (thecall && + !(thecall->msg->header.prog == client->msg.header.prog && + thecall->msg->header.vers == client->msg.header.vers && + thecall->msg->header.serial == client->msg.header.serial)) + thecall = thecall->next; + + if (!thecall) { + virNetError(VIR_ERR_RPC, + _("no call waiting for reply with prog %d vers %d serial %d"), + client->msg.header.prog, client->msg.header.vers, client->msg.header.serial); + return -1; + } + + memcpy(thecall->msg->buffer, client->msg.buffer, sizeof(client->msg.buffer)); + memcpy(&thecall->msg->header, &client->msg.header, sizeof(client->msg.header)); + thecall->msg->bufferLength = client->msg.bufferLength; + thecall->msg->bufferOffset = client->msg.bufferOffset; + + thecall->mode = VIR_NET_CLIENT_MODE_COMPLETE; + + return 0; +} + +static int virNetClientCallDispatchMessage(virNetClientPtr client) +{ + int i; + virNetClientProgramPtr prog = NULL; + + for (i = 0 ; i < client->nprograms ; i++) { + if (virNetClientProgramMatches(client->programs[i], + &client->msg)) { + prog = client->programs[i]; + break; + } + } + if (!prog) { + VIR_DEBUG("No program found for event with prog=%d vers=%d", + client->msg.header.prog, client->msg.header.vers); + return -1; + } + + virNetClientProgramDispatch(prog, client, &client->msg); + + return 0; +} + +static int virNetClientCallDispatchStream(virNetClientPtr client ATTRIBUTE_UNUSED) +{ +#if 0 + struct private_stream_data *privst; + virNetClientCallPtr thecall; + + /* Try and find a matching stream */ + privst = client->streams; + while (privst && + privst->serial != hdr->serial && + privst->proc_nr != hdr->proc) + privst = privst->next; + + if (!privst) { + VIR_DEBUG("No registered stream matching serial=%d, proc=%d", + hdr->serial, hdr->proc); + return -1; + } + + /* See if there's also a (optional) call waiting for this reply */ + thecall = client->waitDispatch; + while (thecall && + thecall->serial != hdr->serial) + thecall = thecall->next; + + + /* Status is either REMOTE_OK (meaning that what follows is a ret + * structure), or REMOTE_ERROR (and what follows is a remote_error + * structure). + */ + switch (hdr->status) { + case REMOTE_CONTINUE: { + int avail = privst->incomingLength - privst->incomingOffset; + int need = client->bufferLength - client->bufferOffset; + VIR_DEBUG0("Got a stream data packet"); + + /* XXX flag stream as complete somwhere if need==0 */ + + if (need > avail) { + int extra = need - avail; + if (VIR_REALLOC_N(privst->incoming, + privst->incomingLength + extra) < 0) { + VIR_DEBUG0("Out of memory handling stream data"); + return -1; + } + privst->incomingLength += extra; + } + + memcpy(privst->incoming + privst->incomingOffset, + client->buffer + client->bufferOffset, + client->bufferLength - client->bufferOffset); + privst->incomingOffset += (client->bufferLength - client->bufferOffset); + + if (thecall && thecall->want_reply) { + VIR_DEBUG("Got sync data packet offset=%d", privst->incomingOffset); + thecall->mode = REMOTE_MODE_COMPLETE; + } else { + VIR_DEBUG("Got aysnc data packet offset=%d", privst->incomingOffset); + remoteStreamEventTimerUpdate(privst); + } + return 0; + } + + case REMOTE_OK: + VIR_DEBUG0("Got a synchronous confirm"); + if (!thecall) { + VIR_DEBUG0("Got unexpected stream finish confirmation"); + return -1; + } + thecall->mode = REMOTE_MODE_COMPLETE; + return 0; + + case REMOTE_ERROR: + if (thecall && thecall->want_reply) { + VIR_DEBUG0("Got a synchronous error"); + /* Give the error straight to this call */ + memset (&thecall->err, 0, sizeof thecall->err); + if (!xdr_remote_error (xdr, &thecall->err)) { + remoteError(VIR_ERR_RPC, "%s", _("unmarshalling remote_error")); + return -1; + } + thecall->mode = REMOTE_MODE_ERROR; + } else { + VIR_DEBUG0("Got a asynchronous error"); + /* No call, so queue the error against the stream */ + if (privst->has_error) { + VIR_DEBUG0("Got unexpected duplicate stream error"); + return -1; + } + privst->has_error = 1; + memset (&privst->err, 0, sizeof privst->err); + if (!xdr_remote_error (xdr, &privst->err)) { + VIR_DEBUG0("Failed to unmarshall error"); + return -1; + } + } + return 0; + + default: + VIR_WARN("Stream with unexpected serial=%d, proc=%d, status=%d", + hdr->serial, hdr->proc, hdr->status); + return -1; + } +#endif + return 0; +} + + +static int +virNetClientCallDispatch(virNetClientPtr client) +{ + if (virNetMessageDecodeHeader(&client->msg) < 0) + return -1; + + switch (client->msg.header.type) { + case VIR_NET_REPLY: /* Normal RPC replies */ + return virNetClientCallDispatchReply(client); + + case VIR_NET_MESSAGE: /* Async notifications */ + return virNetClientCallDispatchMessage(client); + + case VIR_NET_STREAM: /* Stream protocol */ + return virNetClientCallDispatchStream(client); + + default: + virNetError(VIR_ERR_RPC, + _("got unexpected RPC call prog %d vers %d proc %d type %d"), + client->msg.header.prog, client->msg.header.vers, + client->msg.header.proc, client->msg.header.type); + return -1; + } +} + + +static ssize_t +virNetClientIOWriteBuffer(virNetClientPtr client, + const char *bytes, size_t len) +{ + ssize_t ret; + + resend: + if (client->tls) + ret = virNetTLSSessionWrite(client->tls, bytes, len); + else + ret = virNetSocketWrite(client->sock, bytes, len); + if (ret < 0) { + if (errno == EINTR) + goto resend; + if (errno == EAGAIN) + return 0; + + virReportSystemError(errno, "%s", _("cannot send data")); + return -1; + } + + return ret; +} + + +static ssize_t +virNetClientIOReadBuffer(virNetClientPtr client, + char *bytes, size_t len) +{ + size_t ret; + +resend: + if (client->tls) + ret = virNetTLSSessionRead(client->tls, bytes, len); + else + ret = virNetSocketRead(client->sock, bytes, len); + if (ret <= 0) { + if (ret == -1) { + if (errno == EINTR) + goto resend; + if (errno == EAGAIN) + return 0; + + virReportSystemError(errno, "%s", + _("cannot recv data")); + } else { + virNetError(VIR_ERR_SYSTEM_ERROR, "%s", + _("server closed connection")); + } + return -1; + } + + return ret; +} + + +static ssize_t +virNetClientIOWriteMessage(virNetClientPtr client, + virNetClientCallPtr thecall) +{ +#if HAVE_SASL + if (client->sasl) { + const char *output; + size_t outputlen; + ssize_t ret; + + if (!client->saslEncoded) { + if (virNetClientSaslContextEncode(client->sasl, + thecall->msg->buffer + thecall->msg->bufferOffset, + thecall->msg->bufferLength - thecall->msg->bufferOffset, + &output, &outputlen) < 0) + return -1; + + client->saslEncoded = output; + client->saslEncodedLength = outputlen; + client->saslEncodedOffset = 0; + + thecall->msg->bufferOffset = thecall->msg->bufferLength; + } + + ret = virNetClientIOWriteBuffer(client, + client->saslEncoded + client->saslEncodedOffset, + client->saslEncodedLength - client->saslEncodedOffset); + if (ret < 0) + return ret; + client->saslEncodedOffset += ret; + + if (client->saslEncodedOffset == client->saslEncodedLength) { + client->saslEncoded = NULL; + client->saslEncodedOffset = client->saslEncodedLength = 0; + if (thecall->expectReply) + thecall->mode = VIR_NET_CLIENT_MODE_WAIT_RX; + else + thecall->mode = VIR_NET_CLIENT_MODE_COMPLETE; + } + } else { +#endif + ssize_t ret; + ret = virNetClientIOWriteBuffer(client, + thecall->msg->buffer + thecall->msg->bufferOffset, + thecall->msg->bufferLength - thecall->msg->bufferOffset); + if (ret < 0) + return ret; + thecall->msg->bufferOffset += ret; + + if (thecall->msg->bufferOffset == thecall->msg->bufferLength) { + thecall->msg->bufferOffset = thecall->msg->bufferLength = 0; + if (thecall->expectReply) + thecall->mode = VIR_NET_CLIENT_MODE_WAIT_RX; + else + thecall->mode = VIR_NET_CLIENT_MODE_COMPLETE; + } +#if HAVE_SASL + } +#endif + return 0; +} + + +static ssize_t +virNetClientIOHandleOutput(virNetClientPtr client) +{ + virNetClientCallPtr thecall = client->waitDispatch; + + while (thecall && + thecall->mode != VIR_NET_CLIENT_MODE_WAIT_TX) + thecall = thecall->next; + + if (!thecall) + return -1; /* Shouldn't happen, but you never know... */ + + while (thecall) { + ssize_t ret = virNetClientIOWriteMessage(client, thecall); + if (ret < 0) + return ret; + + if (thecall->mode == VIR_NET_CLIENT_MODE_WAIT_TX) + return 0; /* Blocking write, to back to event loop */ + + thecall = thecall->next; + } + + return 0; /* No more calls to send, all done */ +} + +static ssize_t +virNetClientIOReadMessage(virNetClientPtr client) +{ + size_t wantData; + + /* Start by reading length word */ + if (client->msg.bufferLength == 0) + client->msg.bufferLength = 4; + + wantData = client->msg.bufferLength - client->msg.bufferOffset; + +#if HAVE_SASL + if (client->sasl) { + if (client->saslDecoded == NULL) { + char encoded[8192]; + ssize_t ret; + ret = virNetClientIOReadBuffer(client, encoded, sizeof(encoded)); + if (ret < 0) + return -1; + if (ret == 0) + return 0; + + if (virNetClientSaslContextDecode(client->sasl, + encoded, + ret, + &client->saslDecoded, + &client->saslDecodedLength) < 0) + return -1; + + client->saslDecodedOffset = 0; + } + + if ((client->saslDecodedLength - client->saslDecodedOffset) < wantData) + wantData = (client->saslDecodedLength - client->saslDecodedOffset); + + memcpy(client->msg.buffer + client->msg.bufferOffset, + client->saslDecoded + client->saslDecodedOffset, + wantData); + client->saslDecodedOffset += wantData; + client->msg.bufferOffset += wantData; + if (client->saslDecodedOffset == client->saslDecodedLength) { + client->saslDecodedOffset = client->saslDecodedLength = 0; + client->saslDecoded = NULL; + } + + return wantData; + } else { +#endif + ssize_t ret; + + ret = virNetClientIOReadBuffer(client, + client->msg.buffer + client->msg.bufferOffset, + wantData); + if (ret < 0) + return -1; + if (ret == 0) + return 0; + + client->msg.bufferOffset += ret; + + return ret; +#if HAVE_SASL + } +#endif +} + + +static ssize_t +virNetClientIOHandleInput(virNetClientPtr client) +{ + /* Read as much data as is available, until we get + * EAGAIN + */ + for (;;) { + ssize_t ret = virNetClientIOReadMessage(client); + + if (ret < 0) + return -1; + if (ret == 0) + return 0; /* Blocking on read */ + + /* Check for completion of our goal */ + if (client->msg.bufferOffset == client->msg.bufferLength) { + if (client->msg.bufferOffset == 4) { + ret = virNetMessageDecodeLength(&client->msg); + if (ret < 0) + return -1; + + /* + * We'll carry on around the loop to immediately + * process the message body, because it has probably + * already arrived. Worst case, we'll get EAGAIN on + * next iteration. + */ + } else { + ret = virNetClientCallDispatch(client); + client->msg.bufferOffset = client->msg.bufferLength = 0; + /* + * We've completed one call, so return even + * though there might still be more data on + * the wire. We need to actually let the caller + * deal with this arrived message to keep good + * response, and also to correctly handle EOF. + */ + return ret; + } + } + } +} + + +/* + * Process all calls pending dispatch/receive until we + * get a reply to our own call. Then quit and pass the buck + * to someone else. + */ +static int virNetClientIOEventLoop(virNetClientPtr client, + virNetClientCallPtr thiscall) +{ + struct pollfd fds[2]; + int ret; + + fds[0].fd = virNetSocketFD(client->sock); + fds[1].fd = client->wakeupReadFD; + + for (;;) { + virNetClientCallPtr tmp = client->waitDispatch; + virNetClientCallPtr prev; + char ignore; +#ifdef HAVE_PTHREAD_SIGMASK + sigset_t oldmask, blockedsigs; +#endif + + fds[0].events = fds[0].revents = 0; + fds[1].events = fds[1].revents = 0; + + fds[1].events = POLLIN; + while (tmp) { + if (tmp->mode == VIR_NET_CLIENT_MODE_WAIT_RX) + fds[0].events |= POLLIN; + if (tmp->mode == VIR_NET_CLIENT_MODE_WAIT_TX) + fds[0].events |= POLLOUT; + + tmp = tmp->next; + } + +#if 0 + XXX + if (client->streams) + fds[0].events |= POLLIN; +#endif + + /* Release lock while poll'ing so other threads + * can stuff themselves on the queue */ + virNetClientUnlock(client); + + /* Block SIGWINCH from interrupting poll in curses programs, + * then restore the original signal mask again immediately + * after the call (RHBZ#567931). Same for SIGCHLD and SIGPIPE + * at the suggestion of Paolo Bonzini and Daniel Berrange. + */ +#ifdef HAVE_PTHREAD_SIGMASK + sigemptyset (&blockedsigs); + sigaddset (&blockedsigs, SIGWINCH); + sigaddset (&blockedsigs, SIGCHLD); + sigaddset (&blockedsigs, SIGPIPE); + ignore_value(pthread_sigmask(SIG_BLOCK, &blockedsigs, &oldmask)); +#endif + + repoll: + ret = poll(fds, ARRAY_CARDINALITY(fds), -1); + if (ret < 0 && errno == EAGAIN) + goto repoll; + +#ifdef HAVE_PTHREAD_SIGMASK + ignore_value(pthread_sigmask(SIG_SETMASK, &oldmask, NULL)); +#endif + + virNetClientLock(client); + + if (fds[1].revents) { + VIR_DEBUG0("Woken up from poll by other thread"); + if (saferead(client->wakeupReadFD, &ignore, sizeof(ignore)) != sizeof(ignore)) { + virReportSystemError(errno, "%s", + _("read on wakeup fd failed")); + goto error; + } + } + + if (ret < 0) { + if (errno == EWOULDBLOCK) + continue; + virReportSystemError(errno, + "%s", _("poll on socket failed")); + goto error; + } + + if (fds[0].revents & POLLOUT) { + if (virNetClientIOHandleOutput(client) < 0) + goto error; + } + + if (fds[0].revents & POLLIN) { + if (virNetClientIOHandleInput(client) < 0) + goto error; + } + + /* Iterate through waiting threads and if + * any are complete then tell 'em to wakeup + */ + tmp = client->waitDispatch; + prev = NULL; + while (tmp) { + if (tmp != thiscall && + tmp->mode == VIR_NET_CLIENT_MODE_COMPLETE) { + /* Take them out of the list */ + if (prev) + prev->next = tmp->next; + else + client->waitDispatch = tmp->next; + + /* And wake them up.... + * ...they won't actually wakeup until + * we release our mutex a short while + * later... + */ + VIR_DEBUG("Waking up sleep %p %p", tmp, client->waitDispatch); + virCondSignal(&tmp->cond); + } + prev = tmp; + tmp = tmp->next; + } + + /* Now see if *we* are done */ + if (thiscall->mode == VIR_NET_CLIENT_MODE_COMPLETE) { + /* We're at head of the list already, so + * remove us + */ + client->waitDispatch = thiscall->next; + VIR_DEBUG("Giving up the buck %p %p", thiscall, client->waitDispatch); + /* See if someone else is still waiting + * and if so, then pass the buck ! */ + if (client->waitDispatch) { + VIR_DEBUG("Passing the buck to %p", client->waitDispatch); + virCondSignal(&client->waitDispatch->cond); + } + return 0; + } + + + if (fds[0].revents & (POLLHUP | POLLERR)) { + virNetError(VIR_ERR_INTERNAL_ERROR, "%s", + _("received hangup / error event on socket")); + goto error; + } + } + + +error: + client->waitDispatch = thiscall->next; + VIR_DEBUG("Giving up the buck due to I/O error %p %p", thiscall, client->waitDispatch); + /* See if someone else is still waiting + * and if so, then pass the buck ! */ + if (client->waitDispatch) { + VIR_DEBUG("Passing the buck to %p", client->waitDispatch); + virCondSignal(&client->waitDispatch->cond); + } + return -1; +} + + +/* + * This function sends a message to remote server and awaits a reply + * + * NB. This does not free the args structure (not desirable, since you + * often want this allocated on the stack or else it contains strings + * which come from the user). It does however free any intermediate + * results, eg. the error structure if there is one. + * + * NB(2). Make sure to memset (&ret, 0, sizeof ret) before calling, + * else Bad Things will happen in the XDR code. + * + * NB(3) You must have the client lock before calling this + * + * NB(4) This is very complicated. Multiple threads are allowed to + * use the client for RPC at the same time. Obviously only one of + * them can. So if someone's using the socket, other threads are put + * to sleep on condition variables. The existing thread may completely + * send & receive their RPC call/reply while they're asleep. Or it + * may only get around to dealing with sending the call. Or it may + * get around to neither. So upon waking up from slumber, the other + * thread may or may not have more work todo. + * + * We call this dance 'passing the buck' + * + * http://en.wikipedia.org/wiki/Passing_the_buck + * + * "Buck passing or passing the buck is the action of transferring + * responsibility or blame unto another person. It is also used as + * a strategy in power politics when the actions of one country/ + * nation are blamed on another, providing an opportunity for war." + * + * NB(5) Don't Panic! + */ +static int virNetClientIO(virNetClientPtr client, + virNetClientCallPtr thiscall) +{ + int rv = -1; + + VIR_DEBUG("program=%u version=%u serial=%u proc=%d type=%d length=%d dispatach=%p", + thiscall->msg->header.prog, + thiscall->msg->header.vers, + thiscall->msg->header.serial, + thiscall->msg->header.proc, + thiscall->msg->header.type, + thiscall->msg->bufferLength, + client->waitDispatch); + + /* Check to see if another thread is dispatching */ + if (client->waitDispatch) { + /* Stick ourselves on the end of the wait queue */ + virNetClientCallPtr tmp = client->waitDispatch; + char ignore = 1; + while (tmp && tmp->next) + tmp = tmp->next; + if (tmp) + tmp->next = thiscall; + else + client->waitDispatch = thiscall; + + /* Force other thread to wakeup from poll */ + if (safewrite(client->wakeupSendFD, &ignore, sizeof(ignore)) != sizeof(ignore)) { + if (tmp) + tmp->next = NULL; + else + client->waitDispatch = NULL; + virReportSystemError(errno, "%s", + _("failed to wake up polling thread")); + return -1; + } + + VIR_DEBUG("Going to sleep %p %p", client->waitDispatch, thiscall); + /* Go to sleep while other thread is working... */ + if (virCondWait(&thiscall->cond, &client->lock) < 0) { + if (client->waitDispatch == thiscall) { + client->waitDispatch = thiscall->next; + } else { + tmp = client->waitDispatch; + while (tmp && tmp->next && + tmp->next != thiscall) { + tmp = tmp->next; + } + if (tmp && tmp->next == thiscall) + tmp->next = thiscall->next; + } + virNetError(VIR_ERR_INTERNAL_ERROR, "%s", + _("failed to wait on condition")); + return -1; + } + + VIR_DEBUG("Wokeup from sleep %p %p", client->waitDispatch, thiscall); + /* Two reasons we can be woken up + * 1. Other thread has got our reply ready for us + * 2. Other thread is all done, and it is our turn to + * be the dispatcher to finish waiting for + * our reply + */ + if (thiscall->mode == VIR_NET_CLIENT_MODE_COMPLETE) { + rv = 0; + /* + * We avoided catching the buck and our reply is ready ! + * We've already had 'thiscall' removed from the list + * so just need to (maybe) handle errors & free it + */ + goto cleanup; + } + + /* Grr, someone passed the buck onto us ... */ + + } else { + /* We're first to catch the buck */ + client->waitDispatch = thiscall; + } + + VIR_DEBUG("We have the buck %p %p", client->waitDispatch, thiscall); + /* + * The buck stops here! + * + * At this point we're about to own the dispatch + * process... + */ + + /* + * Avoid needless wake-ups of the event loop in the + * case where this call is being made from a different + * thread than the event loop. These wake-ups would + * cause the event loop thread to be blocked on the + * mutex for the duration of the call + */ + virNetSocketUpdateIOCallback(client->sock, 0); + + rv = virNetClientIOEventLoop(client, thiscall); + + virNetSocketUpdateIOCallback(client->sock, VIR_EVENT_HANDLE_READABLE); + +cleanup: + VIR_DEBUG("All done with our call %p %p %d", client->waitDispatch, thiscall, rv); + return rv; +} + + +void virNetClientIncomingEvent(virNetSocketPtr sock, + int events, + void *opaque) +{ + virNetClientPtr client = opaque; + + virNetClientLock(client); + + /* This should be impossible, but it doesn't hurt to check */ + if (client->waitDispatch) + goto done; + + VIR_DEBUG("Event fired %p %d", sock, events); + + if (events & (VIR_EVENT_HANDLE_HANGUP | VIR_EVENT_HANDLE_ERROR)) { + VIR_DEBUG("%s : VIR_EVENT_HANDLE_HANGUP or " + "VIR_EVENT_HANDLE_ERROR encountered", __FUNCTION__); + virNetSocketRemoveIOCallback(sock); + goto done; + } + + if (virNetClientIOHandleInput(client) < 0) + VIR_DEBUG0("Something went wrong during async message processing"); + +done: + virNetClientUnlock(client); +} + + +int virNetClientSend(virNetClientPtr client, + virNetMessagePtr msg, + bool expectReply) +{ + virNetClientCallPtr call; + int ret = -1; + + if (VIR_ALLOC(call) < 0) { + virReportOOMError(); + return -1; + } + + virNetClientLock(client); + + if (virCondInit(&call->cond) < 0) { + virNetError(VIR_ERR_INTERNAL_ERROR, "%s", + _("cannot initialize condition variable")); + goto cleanup; + } + + call->mode = VIR_NET_CLIENT_MODE_WAIT_TX; + call->msg = msg; + call->expectReply = expectReply; + + ret = virNetClientIO(client, call); + +cleanup: + VIR_FREE(call); + virNetClientUnlock(client); + return ret; +} diff --git a/src/rpc/virnetclient.h b/src/rpc/virnetclient.h new file mode 100644 index 0000000..cd6a20f --- /dev/null +++ b/src/rpc/virnetclient.h @@ -0,0 +1,60 @@ + + +#ifndef __VIR_NET_CLIENT_H__ +#define __VIR_NET_CLIENT_H__ + +#include <stdbool.h> + +#include "virnettlscontext.h" +#include "virnetmessage.h" +#if HAVE_SASL +#include "virnetclientsaslcontext.h" +#endif +#include "virnetclientprogram.h" + + +virNetClientPtr virNetClientNewUNIX(const char *path, + bool spawnDaemon, + const char *daemon); + +virNetClientPtr virNetClientNewTCP(const char *nodename, + const char *service); + +virNetClientPtr virNetClientNewSSH(const char *nodename, + const char *service, + const char *binary, + const char *username, + bool noTTY, + const char *netcat, + const char *path); + +virNetClientPtr virNetClientNewCommand(const char **cmdargv, + const char **cmdenv); + +void virNetClientRef(virNetClientPtr client); + +int virNetClientAddProgram(virNetClientPtr client, + virNetClientProgramPtr prog); + +int virNetClientSend(virNetClientPtr client, + virNetMessagePtr msg, + bool expectReply); + +#if HAVE_SASL +void virNetClientSetSASLContext(virNetClientPtr client, + virNetClientSaslContextPtr ctxt); +#endif + +int virNetClientSetTLSSession(virNetClientPtr client, + virNetTLSContextPtr tls); + +bool virNetClientIsEncrypted(virNetClientPtr client); + +const char *virNetClientLocalAddrString(virNetClientPtr client); +const char *virNetClientRemoteAddrString(virNetClientPtr client); + +int virNetClientGetTLSKeySize(virNetClientPtr client); + +void virNetClientFree(virNetClientPtr client); + +#endif /* __VIR_NET_CLIENT_H__ */ diff --git a/src/rpc/virnetclientprogram.c b/src/rpc/virnetclientprogram.c new file mode 100644 index 0000000..eb918f0 --- /dev/null +++ b/src/rpc/virnetclientprogram.c @@ -0,0 +1,258 @@ + +#include <config.h> + +#include "virnetclientprogram.h" +#include "virnetclient.h" +#include "virnetprotocol.h" + +#include "memory.h" +#include "virterror_internal.h" +#include "logging.h" + +#define VIR_FROM_THIS VIR_FROM_RPC + +#define virNetError(code, ...) \ + virReportErrorHelper(NULL, VIR_FROM_RPC, code, __FILE__, \ + __FUNCTION__, __LINE__, __VA_ARGS__) + +struct _virNetClientProgram { + int refs; + + unsigned program; + unsigned version; + virNetClientProgramEventPtr events; + size_t nevents; + void *eventOpaque; + virNetClientProgramErrorHanderPtr err; +}; + +virNetClientProgramPtr virNetClientProgramNew(unsigned program, + unsigned version, + virNetClientProgramEventPtr events, + size_t nevents, + void *eventOpaque, + virNetClientProgramErrorHanderPtr err) +{ + virNetClientProgramPtr prog; + + if (VIR_ALLOC(prog) < 0) { + virReportOOMError(); + return NULL; + } + + prog->refs = 1; + prog->program = program; + prog->version = version; + prog->events = events; + prog->nevents = nevents; + prog->eventOpaque = eventOpaque; + prog->err = err; + + return prog; +} + + +void virNetClientProgramRef(virNetClientProgramPtr prog) +{ + prog->refs++; +} + + +void virNetClientProgramFree(virNetClientProgramPtr prog) +{ + if (!prog) + return; + + prog->refs--; + if (prog->refs > 0) + return; + + VIR_FREE(prog); +} + + +int virNetClientProgramMatches(virNetClientProgramPtr prog, + virNetMessagePtr msg) +{ + if (prog->program == msg->header.prog && + prog->version == msg->header.vers) + return 1; + return 0; +} + + +static int virNetClientProgramDispatchError(virNetClientProgramPtr prog, + virNetMessagePtr msg) +{ + char *err; + int ret = -1; + + if (VIR_ALLOC_N(err, prog->err->len) < 0) { + virReportOOMError(); + return -1; + } + + if (virNetMessageDecodePayload(msg, prog->err->filter, err) < 0) + goto cleanup; + + prog->err->func(prog, err); + + ret = 0; + +cleanup: + VIR_FREE(err); + return ret; +} + +static virNetClientProgramEventPtr virNetClientProgramGetEvent(virNetClientProgramPtr prog, + int procedure) +{ + int i; + + for (i = 0 ; i < prog->nevents ; i++) { + if (prog->events[i].proc == procedure) + return &prog->events[i]; + } + + return NULL; +} + + +int virNetClientProgramDispatch(virNetClientProgramPtr prog, + virNetClientPtr client, + virNetMessagePtr msg) +{ + virNetClientProgramEventPtr event; + char *evdata; + + VIR_DEBUG("prog=%d ver=%d type=%d status=%d serial=%d proc=%d", + msg->header.prog, msg->header.vers, msg->header.type, + msg->header.status, msg->header.serial, msg->header.proc); + + /* Check version, etc. */ + if (msg->header.prog != prog->program) { + VIR_ERROR(_("program mismatch in event (actual %x, expected %x)"), + msg->header.prog, prog->program); + return -1; + } + + if (msg->header.vers != prog->version) { + VIR_ERROR(_("version mismatch in event (actual %x, expected %x)"), + msg->header.vers, prog->version); + return -1; + } + + if (msg->header.status != VIR_NET_OK) { + VIR_ERROR(_("status mismatch in event (actual %x, expected %x)"), + msg->header.status, VIR_NET_OK); + return -1; + } + + if (msg->header.type != VIR_NET_MESSAGE) { + VIR_ERROR(_("type mismatch in event (actual %x, expected %x)"), + msg->header.type, VIR_NET_MESSAGE); + return -1; + } + + event = virNetClientProgramGetEvent(prog, msg->header.proc); + + if (!event) { + VIR_ERROR(_("No event expected with procedure %x"), + msg->header.proc); + return -1; + } + + if (VIR_ALLOC_N(evdata, event->msg_len) < 0) { + virReportOOMError(); + return -1; + } + + if (virNetMessageDecodePayload(msg, event->msg_filter, evdata) < 0) + goto cleanup; + + event->func(prog, client, &evdata, prog->eventOpaque); + + xdr_free(event->msg_filter, evdata); + +cleanup: + VIR_FREE(evdata); + return 0; +} + + +int virNetClientProgramCall(virNetClientProgramPtr prog, + virNetClientPtr client, + unsigned serial, + int proc, + xdrproc_t args_filter, void *args, + xdrproc_t ret_filter, void *ret) +{ + virNetMessagePtr msg; + + if (VIR_ALLOC(msg) < 0) { + virReportOOMError(); + return -1; + } + + msg->header.prog = prog->program; + msg->header.vers = prog->version; + msg->header.status = VIR_NET_OK; + msg->header.type = VIR_NET_CALL; + msg->header.serial = serial; + msg->header.proc = proc; + + if (virNetMessageEncodeHeader(msg) < 0) + goto error; + + if (virNetMessageEncodePayload(msg, args_filter, args) < 0) + goto error; + + if (virNetClientSend(client, msg, true) < 0) + goto error; + + /* None of these 3 should ever happen here, because + * virNetClientSend should have validated the reply, + * but it doesn't hurt to check again. + */ + if (msg->header.type != VIR_NET_REPLY) { + virNetError(VIR_ERR_INTERNAL_ERROR, + _("Unexpected message type %d"), msg->header.type); + goto error; + } + if (msg->header.proc != proc) { + virNetError(VIR_ERR_INTERNAL_ERROR, + _("Unexpected message proc %d != %d"), + msg->header.proc, proc); + goto error; + } + if (msg->header.serial != serial) { + virNetError(VIR_ERR_INTERNAL_ERROR, + _("Unexpected message serial %d != %d"), + msg->header.serial, serial); + goto error; + } + + switch (msg->header.status) { + case VIR_NET_OK: + if (virNetMessageDecodePayload(msg, ret_filter, ret) < 0) + goto error; + break; + + case VIR_NET_ERROR: + virNetClientProgramDispatchError(prog, msg); + goto error; + + default: + virNetError(VIR_ERR_RPC, + _("Unexpected message status %d"), msg->header.status); + goto error; + } + + VIR_FREE(msg); + + return 0; + +error: + VIR_FREE(msg); + return -1; +} diff --git a/src/rpc/virnetclientprogram.h b/src/rpc/virnetclientprogram.h new file mode 100644 index 0000000..5a5b937 --- /dev/null +++ b/src/rpc/virnetclientprogram.h @@ -0,0 +1,71 @@ + +#ifndef __VIR_NET_CLIENT_PROGRAM_H__ +#define __VIR_NET_CLIENT_PROGRAM_H__ + +#include <rpc/types.h> +#include <rpc/xdr.h> + +#include "virnetmessage.h" + +typedef struct _virNetClient virNetClient; +typedef virNetClient *virNetClientPtr; + +typedef struct _virNetClientProgram virNetClientProgram; +typedef virNetClientProgram *virNetClientProgramPtr; + +typedef struct _virNetClientProgramEvent virNetClientProgramEvent; +typedef virNetClientProgramEvent *virNetClientProgramEventPtr; + +typedef struct _virNetClientProgramErrorHandler virNetClientProgramErrorHander; +typedef virNetClientProgramErrorHander *virNetClientProgramErrorHanderPtr; + +typedef int (*virNetClientProgramErrorFunc)(virNetClientProgramPtr prog, + void *rerr); + +struct _virNetClientProgramErrorHandler { + virNetClientProgramErrorFunc func; + size_t len; + xdrproc_t filter; +}; + + +typedef void (*virNetClientProgramDispatchFunc)(virNetClientProgramPtr prog, + virNetClientPtr client, + void *msg, + void *opaque); + +struct _virNetClientProgramEvent { + int proc; + virNetClientProgramDispatchFunc func; + size_t msg_len; + xdrproc_t msg_filter; +}; + +virNetClientProgramPtr virNetClientProgramNew(unsigned program, + unsigned version, + virNetClientProgramEventPtr events, + size_t nevents, + void *eventOpaque, + virNetClientProgramErrorHanderPtr err); + +void virNetClientProgramRef(virNetClientProgramPtr prog); + +void virNetClientProgramFree(virNetClientProgramPtr prog); + +int virNetClientProgramMatches(virNetClientProgramPtr prog, + virNetMessagePtr msg); + +int virNetClientProgramDispatch(virNetClientProgramPtr prog, + virNetClientPtr client, + virNetMessagePtr msg); + +int virNetClientProgramCall(virNetClientProgramPtr prog, + virNetClientPtr client, + unsigned serial, + int proc, + xdrproc_t args_filter, void *args, + xdrproc_t ret_filter, void *ret); + + + +#endif /* __VIR_NET_CLIENT_PROGRAM_H__ */ diff --git a/src/rpc/virnetclientsaslcontext.c b/src/rpc/virnetclientsaslcontext.c new file mode 100644 index 0000000..757cd72 --- /dev/null +++ b/src/rpc/virnetclientsaslcontext.c @@ -0,0 +1,246 @@ + +#include <config.h> + +#include "virnetclientsaslcontext.h" + +#include "virterror_internal.h" +#include "memory.h" +#include "logging.h" + +#define VIR_FROM_THIS VIR_FROM_RPC + +#define virNetError(code, ...) \ + virReportErrorHelper(NULL, VIR_FROM_RPC, code, __FILE__, \ + __FUNCTION__, __LINE__, __VA_ARGS__) + + +struct _virNetClientSaslContext { + sasl_conn_t *conn; + int refs; +}; + +virNetClientSaslContextPtr virNetClientSaslContextNew(const char *service, + const char *hostname, + const char *localAddr, + const char *remoteAddr, + const sasl_callback_t *cbs) +{ + virNetClientSaslContextPtr sasl = NULL; + int err; + + err = sasl_client_init(NULL); + if (err != SASL_OK) { + virNetError(VIR_ERR_AUTH_FAILED, + _("failed to initialize SASL library: %d (%s)"), + err, sasl_errstring(err, NULL, NULL)); + goto cleanup; + } + + if (VIR_ALLOC(sasl) < 0) { + virReportOOMError(); + goto cleanup; + } + + sasl->refs = 1; + + err = sasl_client_new(service, + hostname, + localAddr, + remoteAddr, + cbs, + SASL_SUCCESS_DATA, + &sasl->conn); + if (err != SASL_OK) { + virNetError(VIR_ERR_AUTH_FAILED, + _("Failed to create SASL client context: %d (%s)"), + err, sasl_errstring(err, NULL, NULL)); + goto cleanup; + } + + return sasl; + +cleanup: + virNetClientSaslContextFree(sasl); + return NULL; +} + +void virNetClientSaslContextRef(virNetClientSaslContextPtr sasl) +{ + sasl->refs++; +} + +int virNetClientSaslContextExtKeySize(virNetClientSaslContextPtr sasl, + int ssf) +{ + int err; + + err = sasl_setprop(sasl->conn, SASL_SSF_EXTERNAL, &ssf); + if (err != SASL_OK) { + virNetError(VIR_ERR_INTERNAL_ERROR, + _("cannot set external SSF %d (%s)"), + err, sasl_errstring(err, NULL, NULL)); + return -1; + } + return 0; +} + +int virNetClientSaslContextGetKeySize(virNetClientSaslContextPtr sasl) +{ + int err; + int ssf; + const void *val; + err = sasl_getprop(sasl->conn, SASL_SSF, &val); + if (err != SASL_OK) { + virNetError(VIR_ERR_AUTH_FAILED, + _("cannot query SASL ssf on connection %d (%s)"), + err, sasl_errstring(err, NULL, NULL)); + return -1; + } + ssf = *(const int *)val; + return ssf; +} + +int virNetClientSaslContextSecProps(virNetClientSaslContextPtr sasl, + int minSSF, + int maxSSF, + bool allowAnonymous) +{ + sasl_security_properties_t secprops; + int err; + + memset (&secprops, 0, sizeof secprops); + + secprops.min_ssf = minSSF; + secprops.max_ssf = maxSSF; + secprops.maxbufsize = 100000; + secprops.security_flags = allowAnonymous ? 0 : + SASL_SEC_NOANONYMOUS | SASL_SEC_NOPLAINTEXT; + + err = sasl_setprop(sasl->conn, SASL_SEC_PROPS, &secprops); + if (err != SASL_OK) { + virNetError(VIR_ERR_INTERNAL_ERROR, + _("cannot set security props %d (%s)"), + err, sasl_errstring(err, NULL, NULL)); + return -1; + } + + return 0; +} + + +int virNetClientSaslContextStart(virNetClientSaslContextPtr sasl, + const char *mechlist, + sasl_interact_t **prompt_need, + const char **clientout, + size_t *clientoutlen, + const char **mech) +{ + int err = sasl_client_start(sasl->conn, + mechlist, + prompt_need, + clientout, + (unsigned *)clientoutlen, + mech); + + switch (err) { + case SASL_OK: + return VIR_NET_CLIENT_SASL_COMPLETE; + case SASL_CONTINUE: + return VIR_NET_CLIENT_SASL_CONTINUE; + case SASL_INTERACT: + return VIR_NET_CLIENT_SASL_INTERACT; + + default: + virNetError(VIR_ERR_AUTH_FAILED, + _("Failed to start SASL negotiation: %d (%s)"), + err, sasl_errdetail(sasl->conn)); + return -1; + } +} + + +int virNetClientSaslContextStep(virNetClientSaslContextPtr sasl, + const char *serverin, + size_t serverinlen, + sasl_interact_t **prompt_need, + const char **clientout, + size_t *clientoutlen) +{ + int err = sasl_client_step(sasl->conn, + serverin, + (unsigned)serverinlen, + prompt_need, + clientout, + (unsigned *)clientoutlen); + + + switch (err) { + case SASL_OK: + return VIR_NET_CLIENT_SASL_COMPLETE; + case SASL_CONTINUE: + return VIR_NET_CLIENT_SASL_CONTINUE; + case SASL_INTERACT: + return VIR_NET_CLIENT_SASL_INTERACT; + + default: + virNetError(VIR_ERR_AUTH_FAILED, + _("Failed to start SASL negotiation: %d (%s)"), + err, sasl_errdetail(sasl->conn)); + return -1; + } +} + +ssize_t virNetClientSaslContextEncode(virNetClientSaslContextPtr sasl, + const char *input, + size_t inputLen, + const char **output, + size_t *outputlen) +{ + int err; + err = sasl_encode(sasl->conn, + input, (unsigned)inputLen, + output, (unsigned *)outputlen); + + if (err != SASL_OK) { + virNetError(VIR_ERR_INTERNAL_ERROR, + _("failed to encode SASL data: %d (%s)"), + err, sasl_errstring(err, NULL, NULL)); + return -1; + } + return 0; +} + +ssize_t virNetClientSaslContextDecode(virNetClientSaslContextPtr sasl, + const char *input, + size_t inputLen, + const char **output, + size_t *outputlen) +{ + int err; + err = sasl_decode(sasl->conn, + input, (unsigned)inputLen, + output, (unsigned *)outputlen); + if (err != SASL_OK) { + virNetError(VIR_ERR_INTERNAL_ERROR, + _("failed to decode SASL data: %d (%s)"), + err, sasl_errstring(err, NULL, NULL)); + return -1; + } + return 0; +} + +void virNetClientSaslContextFree(virNetClientSaslContextPtr sasl) +{ + if (!sasl) + return; + + sasl->refs--; + if (sasl->refs > 0) + return; + + if (sasl->conn) + sasl_dispose(&sasl->conn); + + VIR_FREE(sasl); +} + diff --git a/src/rpc/virnetclientsaslcontext.h b/src/rpc/virnetclientsaslcontext.h new file mode 100644 index 0000000..043ae58 --- /dev/null +++ b/src/rpc/virnetclientsaslcontext.h @@ -0,0 +1,66 @@ + +#ifndef __VIR_NET_CLIENT_SASL_CONTEXT_H__ +# define __VIR_NET_CLIENT_SASL_CONTEXT_H__ + +# include <sasl/sasl.h> + +# include <sys/types.h> + +#include "virnetsocket.h" + +typedef struct _virNetClientSaslContext virNetClientSaslContext; +typedef virNetClientSaslContext *virNetClientSaslContextPtr; + +enum { + VIR_NET_CLIENT_SASL_COMPLETE, + VIR_NET_CLIENT_SASL_CONTINUE, + VIR_NET_CLIENT_SASL_INTERACT, +}; + +virNetClientSaslContextPtr virNetClientSaslContextNew(const char *service, + const char *hostname, + const char *localAddr, + const char *remoteAddr, + const sasl_callback_t *cbs); + +void virNetClientSaslContextRef(virNetClientSaslContextPtr sasl); + +int virNetClientSaslContextExtKeySize(virNetClientSaslContextPtr sasl, + int ssf); + +int virNetClientSaslContextGetKeySize(virNetClientSaslContextPtr sasl); + +int virNetClientSaslContextSecProps(virNetClientSaslContextPtr sasl, + int minSSF, + int maxSSF, + bool allowAnonymous); + +int virNetClientSaslContextStart(virNetClientSaslContextPtr sasl, + const char *mechlist, + sasl_interact_t **prompt_need, + const char **clientout, + size_t *clientoutlen, + const char **mech); + +int virNetClientSaslContextStep(virNetClientSaslContextPtr sasl, + const char *serverin, + size_t serverinlen, + sasl_interact_t **prompt_need, + const char **clientout, + size_t *clientoutlen); + +ssize_t virNetClientSaslContextEncode(virNetClientSaslContextPtr sasl, + const char *input, + size_t inputLen, + const char **output, + size_t *outputlen); + +ssize_t virNetClientSaslContextDecode(virNetClientSaslContextPtr sasl, + const char *input, + size_t inputLen, + const char **output, + size_t *outputlen); + +void virNetClientSaslContextFree(virNetClientSaslContextPtr sasl); + +#endif /* __VIR_NET_CLIENT_SASL_CONTEXT_H__ */ -- 1.7.2.3