auditd_plugins/remote/transport.c

/*
 * CDDL HEADER START
 *
 * The contents of this file are subject to the terms of the
 * Common Development and Distribution License (the "License").
 * You may not use this file except in compliance with the License.
 *
 * You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE
 * or http://www.opensolaris.org/os/licensing.
 * See the License for the specific language governing permissions
 * and limitations under the License.
 *
 * When distributing Covered Code, include this CDDL HEADER in each
 * file and include the License file at usr/src/OPENSOLARIS.LICENSE.
 * If applicable, add the following below this CDDL HEADER, with the
 * fields enclosed by brackets "[]" replaced with your own identifying
 * information: Portions Copyright [yyyy] [name of copyright owner]
 *
 * CDDL HEADER END
 */
/*
 * Copyright 2009 Sun Microsystems, Inc.  All rights reserved.
 * Use is subject to license terms.
 *
 * transport layer for audit_remote (handles connection establishment, gss
 * context initialization, message encryption and verification)
 *
 */

#include <assert.h>
#include <audit_plugin.h>
#include <errno.h>
#include <fcntl.h>
#include <gssapi/gssapi.h>
#include <libintl.h>
#include <mtmalloc.h>
#include <netdb.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <strings.h>
#include <syslog.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <unistd.h>
#include <poll.h>
#include <pthread.h>

#include "audit_remote.h"


static int 		sockfd = -1;
static struct hostent	*current_host;
static gss_OID		*current_mech_oid;
static in_port_t 	current_port;
static boolean_t	flush_transq;

static char		*ver_str = "01";	/* supported protocol version */
static char		*ver_str_concat;	/* concat serv/client version */

static gss_ctx_id_t	gss_ctx;
static boolean_t	gss_ctx_initialized;

pthread_t		recv_tid;		/* receiving thread */
static pthread_once_t	recv_once_control = PTHREAD_ONCE_INIT;

extern int		timeout;		/* connection timeout */

extern pthread_mutex_t	plugin_mutex;
transq_hdr_t		transq_hdr;

/*
 * The three locks synchronize the simultaneous actions on top of transmission
 * queue, socket, gss_context.
 */
pthread_mutex_t		transq_lock = PTHREAD_MUTEX_INITIALIZER;
pthread_mutex_t		sock_lock = PTHREAD_MUTEX_INITIALIZER;
pthread_mutex_t		gss_ctx_lock = PTHREAD_MUTEX_INITIALIZER;

/* reset routine synchronization - required by the sending thread */
pthread_mutex_t		reset_lock = PTHREAD_MUTEX_INITIALIZER;
static boolean_t	reset_in_progress;	/* reset routine in progress */

#define	NP_CLOSE	-1		/* notification pipe - close message */
#define	NP_EXIT		-2		/* notification pipe - exit message */
boolean_t		notify_pipe_ready;
int			notify_pipe[2]; /* notif. pipe - receiving thread */

pthread_cond_t		reset_cv = PTHREAD_COND_INITIALIZER;
static close_rsn_t	recv_closure_rsn;

#define	MAX_TOK_LEN	(128 * 1000)	/* max token length we accept (B) */

/* transmission queue helpers */
static void		transq_dequeue(transq_node_t *);
static boolean_t	transq_enqueue(transq_node_t **, gss_buffer_t,
    uint64_t);
static int		transq_retransmit(void);

static boolean_t	init_poll(int);
static void 		do_reset(int *, struct pollfd *, boolean_t);
static void 		do_cleanup(int *, struct pollfd *, boolean_t);

static void		init_recv_record(void);
static void		recv_record();
static int		connect_timeout(int, struct sockaddr *, int);
static int		send_timeout(int, const char *, size_t);
static int		recv_timeout(int, char *, size_t);
static int		send_token(int *, gss_buffer_t);
static int		recv_token(int, gss_buffer_t);


/*
 * report_err() - wrapper, mainly due to enhance the code readability - report
 * error to syslog via call to __audit_syslog().
 */
static void
report_err(char *msg)
{
	__audit_syslog("audit_remote.so", LOG_CONS | LOG_NDELAY, LOG_DAEMON,
	    LOG_ERR, msg);

}


/*
 * report_gss_err() - GSS API error reporting
 */
static void
report_gss_err(char *msg, OM_uint32 maj_stat, OM_uint32 min_stat)
{
	gss_buffer_desc	msg_buf;
	OM_uint32	_min, msg_ctx;
	char		*err_msg;

	/* major stat */
	msg_ctx = 0;
	do {
		(void) gss_display_status(&_min, maj_stat, GSS_C_GSS_CODE,
		    *current_mech_oid, &msg_ctx, &msg_buf);
		(void) asprintf(&err_msg,
		    gettext("GSS API error - %s(%u): %.*s\n"), msg, maj_stat,
		    msg_buf.length, (char *)msg_buf.value);
		if (err_msg != NULL) {
			report_err(err_msg);
			free(err_msg);
		}
		(void) gss_release_buffer(&_min, &msg_buf);
	} while (msg_ctx);

	/* minor stat */
	msg_ctx = 0;
	do {
		(void) gss_display_status(&_min, min_stat, GSS_C_MECH_CODE,
		    *current_mech_oid, &msg_ctx, &msg_buf);
		(void) asprintf(&err_msg,
		    gettext("GSS mech error - %s(%u): %.*s\n"), msg, min_stat,
		    msg_buf.length, (char *)msg_buf.value);
		if (err_msg != NULL) {
			report_err(err_msg);
			free(err_msg);
		}
		(void) gss_release_buffer(&_min, &msg_buf);
	} while (msg_ctx);
}

/*
 * prot_ver_negotiate() - negotiate/acknowledge the protocol version. Currently,
 * there is only one version supported by the plugin - "01".
 * Note: connection must be initiated prior version negotiation
 */
static int
prot_ver_negotiate()
{
	gss_buffer_desc	out_buf, in_buf;
	size_t		ver_str_concat_sz;

	/*
	 * Set the version proposal string - once we support more than
	 * version "01" this part should be extended to solve the concatenation
	 * of supported version identifiers.
	 */
	out_buf.value = (void *)ver_str;
	out_buf.length = strlen((char *)out_buf.value);
	DPRINT((dfile, "Protocol version proposal (size=%d): %.*s\n",
	    out_buf.length, out_buf.length, (char *)out_buf.value));

	if (send_token(&sockfd, &out_buf) < 0) {
		DPRINT((dfile, "Sending protocol version token failed\n"));
		return (-1);
	}

	if (recv_token(sockfd, &in_buf) < 0) {
		DPRINT((dfile, "Receiving protocol version token failed\n"));
		return (-1);
	}

	/*
	 * Verify the sent/received string - memcmp() is sufficient here
	 * because we support only one version and it is represented by
	 * the "01" string. The received version has to be "01" string as well.
	 */
	if (out_buf.length != in_buf.length ||
	    memcmp(out_buf.value, in_buf.value, out_buf.length) != 0) {
		DPRINT((dfile, "Verification of the protocol version strings "
		    "failed [%d:%s][%d:%s]\n", out_buf.length,
		    (char *)out_buf.value, in_buf.length,
		    (char *)in_buf.value));
		free(in_buf.value);
		return (-1);
	}

	/*
	 * Prepare the concatenated client/server version strings later used
	 * as an application_data field in the gss_channel_bindings_struct
	 * structure.
	 */
	ver_str_concat_sz = out_buf.length + in_buf.length + 1;
	ver_str_concat = (char *)calloc(1, ver_str_concat_sz);
	if (ver_str_concat == NULL) {
		report_err(gettext("Memory allocation failed"));
		DPRINT((dfile, "Memory allocation failed: %s\n",
		    strerror(errno)));
		free(in_buf.value);
		return (-1);
	}
	(void) memcpy(ver_str_concat, out_buf.value, out_buf.length);
	(void) memcpy(ver_str_concat + out_buf.length, in_buf.value,
	    in_buf.length);
	DPRINT((dfile, "Concatenated version strings: %s\n", ver_str_concat));

	DPRINT((dfile, "Protocol version agreed.\n"));
	free(in_buf.value);
	return (0);
}

/*
 * sock_prepare() - creates and connects socket. Function returns
 * B_FALSE/B_TRUE on failure/success and sets the err_rsn accordingly to the
 * reason of failure.
 */
static boolean_t
sock_prepare(int *sockfdptr, struct hostent *host, close_rsn_t *err_rsn)
{
	struct sockaddr_storage	addr;
	struct sockaddr_in	*sin;
	struct sockaddr_in6	*sin6;
	size_t			addr_len;
	int			sock;

	DPRINT((dfile, "Creating socket for %s\n", host->h_name));
	bzero(&addr, sizeof (addr));
	addr.ss_family = host->h_addrtype;
	switch (host->h_addrtype) {
	case AF_INET:
		sin = (struct sockaddr_in *)&addr;
		addr_len = sizeof (struct sockaddr_in);
		bcopy(host->h_addr_list[0],
		    &(sin->sin_addr), sizeof (struct in_addr));
		sin->sin_port = current_port;
		break;
	case AF_INET6:
		sin6 = (struct sockaddr_in6 *)&addr;
		addr_len = sizeof (struct sockaddr_in6);
		bcopy(host->h_addr_list[0],
		    &(sin6->sin6_addr), sizeof (struct in6_addr));
		sin6->sin6_port = current_port;
		break;
	default:
		/* unknown address family */
		*err_rsn = RSN_UNKNOWN_AF;
		return (B_FALSE);
	}
	if ((sock = socket(addr.ss_family, SOCK_STREAM, 0)) == -1) {
		*err_rsn = RSN_SOCKET_CREATE;
		return (B_FALSE);
	}
	DPRINT((dfile, "Socket created, fd=%d, connecting..\n", sock));

	if (connect_timeout(sock, (struct sockaddr *)&addr, addr_len)) {
		(void) close(sock);
		*err_rsn = RSN_CONNECTION_CREATE;
		return (B_FALSE);
	}
	*sockfdptr = sock;
	DPRINT((dfile, "Connected to %s via fd=%d\n", host->h_name,
	    *sockfdptr));

	return (B_TRUE);
}

/*
 * establish_context() - establish the client/server GSS context.
 *
 * Note: connection must be established and version negotiated (in plain text)
 * prior to establishing context.
 */
static int
establish_context()
{
	gss_buffer_desc				send_tok, recv_tok, *token_ptr;
	OM_uint32				maj_stat, min_stat;
	OM_uint32				init_sec_min_stat, ret_flags;
	gss_name_t				gss_name;
	char					*gss_svc_name = "audit";
	char					*svc_name;
	size_t					ver_str_concat_len;
	struct gss_channel_bindings_struct	input_chan_bindings;

	/* GSS service name = gss_svc_name + "@" + remote hostname (fqdn) */
	(void) asprintf(&svc_name, "%s@%s", gss_svc_name, current_host->h_name);
	if (svc_name == NULL) {
		report_err(gettext("Cannot allocate service name\n"));
		DPRINT((dfile, "Memory allocation failed: %s\n",
		    strerror(errno)));
		return (-1);
	}
	DPRINT((dfile, "Service name: %s\n", svc_name));

	send_tok.value = svc_name;
	send_tok.length = strlen(svc_name);
	maj_stat = gss_import_name(&min_stat, &send_tok,
	    (gss_OID)GSS_C_NT_HOSTBASED_SERVICE, &gss_name);
	if (maj_stat != GSS_S_COMPLETE) {
		report_gss_err(gettext("initializing context"), maj_stat,
		    min_stat);
		free(svc_name);
		return (-1);
	}
	token_ptr = GSS_C_NO_BUFFER;
	gss_ctx = GSS_C_NO_CONTEXT;

	/* initialize channel binding */
	bzero(&input_chan_bindings, sizeof (input_chan_bindings));
	ver_str_concat_len = strlen(ver_str_concat) + 1;
	input_chan_bindings.application_data.length = ver_str_concat_len;
	input_chan_bindings.application_data.value = ver_str_concat;

	(void) pthread_mutex_lock(&gss_ctx_lock);
	do {
		maj_stat = gss_init_sec_context(&init_sec_min_stat,
		    GSS_C_NO_CREDENTIAL, &gss_ctx, gss_name, *current_mech_oid,
		    GSS_C_MUTUAL_FLAG | GSS_C_REPLAY_FLAG | GSS_C_SEQUENCE_FLAG
		    | GSS_C_CONF_FLAG, 0, &input_chan_bindings, token_ptr,
		    NULL, &send_tok, &ret_flags, NULL);

		if (token_ptr != GSS_C_NO_BUFFER) {
			(void) gss_release_buffer(&min_stat, &recv_tok);
		}

		if (send_tok.length != 0) {
			DPRINT((dfile,
			    "Sending init_sec_context token (size=%d)\n",
			    send_tok.length));
			if (send_token(&sockfd, &send_tok) < 0) {
				free(svc_name);
				(void) gss_release_name(&min_stat, &gss_name);
				(void) pthread_mutex_unlock(&gss_ctx_lock);
				return (-1);
			}
		}
		if (send_tok.value != NULL) {
			free(send_tok.value);	/* freeing svc_name */
			send_tok.value = NULL;
			send_tok.length = 0;
		}

		if (maj_stat != GSS_S_COMPLETE &&
		    maj_stat != GSS_S_CONTINUE_NEEDED) {
			report_gss_err(gettext("initializing context"),
			    maj_stat, init_sec_min_stat);
			if (gss_ctx == GSS_C_NO_CONTEXT) {
				(void) gss_delete_sec_context(&min_stat,
				    &gss_ctx, GSS_C_NO_BUFFER);
			}
			(void) gss_release_name(&min_stat, &gss_name);
			(void) pthread_mutex_unlock(&gss_ctx_lock);
			return (-1);
		}

		if (maj_stat == GSS_S_CONTINUE_NEEDED) {
			DPRINT((dfile, "continue needed... "));
			if (recv_token(sockfd, &recv_tok) < 0) {
				(void) gss_release_name(&min_stat, &gss_name);
				(void) pthread_mutex_unlock(&gss_ctx_lock);
				return (-1);
			}
			token_ptr = &recv_tok;
		}
	} while (maj_stat == GSS_S_CONTINUE_NEEDED);
	(void) gss_release_name(&min_stat, &gss_name);

	DPRINT((dfile, "context established\n"));
	(void) pthread_mutex_unlock(&gss_ctx_lock);
	return (0);
}

/*
 * delete_context() - release GSS context.
 */
static void
delete_context()
{
	OM_uint32	min_stat;

	(void) gss_delete_sec_context(&min_stat, &gss_ctx, GSS_C_NO_BUFFER);
	DPRINT((dfile, "context deleted\n"));
}

/*
 * send_token() - send GSS token over the wire.
 */
static int
send_token(int *fdptr, gss_buffer_t tok)
{
	uint32_t	len;
	uint32_t	lensz;
	char		*out_buf;
	int		fd;

	(void) pthread_mutex_lock(&sock_lock);
	if (*fdptr == -1) {
		(void) pthread_mutex_unlock(&sock_lock);
		DPRINT((dfile, "Socket detected as closed.\n"));
		return (-1);
	}
	fd = *fdptr;

	len = htonl(tok->length);
	lensz = sizeof (len);

	out_buf = (char *)malloc((size_t)(lensz + tok->length));
	if (out_buf == NULL) {
		(void) pthread_mutex_unlock(&sock_lock);
		report_err(gettext("Memory allocation failed"));
		DPRINT((dfile, "Memory allocation failed: %s\n",
		    strerror(errno)));
		return (-1);
	}
	(void) memcpy((void *)out_buf, (void *)&len, lensz);
	(void) memcpy((void *)(out_buf + lensz), (void *)tok->value,
	    tok->length);

	if (send_timeout(fd, out_buf, (lensz + tok->length))) {
		(void) pthread_mutex_unlock(&sock_lock);
		free(out_buf);
		return (-1);
	}

	(void) pthread_mutex_unlock(&sock_lock);
	free(out_buf);
	return (0);
}


/*
 * recv_token() - receive GSS token over the wire.
 */
static int
recv_token(int fd, gss_buffer_t tok)
{
	uint32_t	len;

	if (recv_timeout(fd, (char *)&len, sizeof (len))) {
		return (-1);
	}
	len = ntohl(len);

	/* simple DOS prevention mechanism */
	if (len > MAX_TOK_LEN) {
		report_err(gettext("Indicated invalid token length"));
		DPRINT((dfile, "Indicated token length > %dB\n", MAX_TOK_LEN));
		return (-1);
	}

	tok->value = (char *)malloc(len);
	if (tok->value == NULL) {
		report_err(gettext("Memory allocation failed"));
		DPRINT((dfile, "Memory allocation failed: %s\n",
		    strerror(errno)));
		tok->length = 0;
		return (-1);
	}

	if (recv_timeout(fd, tok->value, len)) {
		free(tok->value);
		tok->value = NULL;
		tok->length = 0;
		return (-1);
	}

	tok->length = len;
	return (0);
}


/*
 * I/O functions
 */

/*
 * connect_timeout() - sets nonblocking I/O on a socket and timeout-connects
 */
static int
connect_timeout(int sockfd, struct sockaddr *name, int namelen)
{
	int			flags;
	struct pollfd		fds;
	int			rc;
	struct sockaddr_storage	addr;
	socklen_t		addr_len = sizeof (addr);


	flags = fcntl(sockfd, F_GETFL, 0);
	if (fcntl(sockfd, F_SETFL, flags | O_NONBLOCK) == -1) {
		return (-1);
	}
	if (connect(sockfd, name, namelen)) {
		if (!(errno == EINTR || errno == EINPROGRESS ||
		    errno == EWOULDBLOCK)) {
			return (-1);
		}
	}
	fds.fd = sockfd;
	fds.events = POLLOUT;
	for (;;) {
		fds.revents = 0;
		rc = poll(&fds, 1, timeout * 1000);
		if (rc == 0) {	/* timeout */
			return (-1);
		} else if (rc < 0) {
			if (errno == EINTR || errno == EAGAIN) {
				continue;
			} else {
				return (-1);
			}
		}
		if (fds.revents) {
			if (getpeername(sockfd, (struct sockaddr *)&addr,
			    &addr_len))
				return (-1);
		} else {
			return (-1);
		}
		return (0);
	}
}

/*
 * send_timeout() - send data (in chunks if needed, each chunk in timeout secs).
 */
static int
send_timeout(int fd, const char *buf, size_t len)
{
	int		bytes;
	struct pollfd	fds;
	int		rc;

	fds.fd = fd;
	fds.events = POLLOUT;

	while (len) {
		fds.revents = 0;
		rc = poll(&fds, 1, timeout * 1000);
		if (rc == 0) {	/* timeout */
			return (-1);
		} else if (rc < 0) {
			if (errno == EINTR || errno == EAGAIN) {
				continue;
			} else {
				return (-1);
			}
		}
		if (!fds.revents) {
			return (-1);
		}

		bytes = write(fd, buf, len);
		if (bytes < 0) {
			if (errno == EINTR) {
				continue;
			} else {
				return (-1);
			}
		} else if (bytes == 0) {	/* eof */
			return (-1);
		}

		len -= bytes;
		buf += bytes;
	}

	return (0);
}

/*
 * recv_timeout() - receive data (in chunks if needed, each chunk in timeout
 * secs). In case the function is called from receiving thread, the function
 * cycles the poll() call in timeout seconds (waits for input from server).
 */
static int
recv_timeout(int fd, char *buf, size_t len)
{
	int		bytes;
	struct pollfd	fds;
	int		rc;

	fds.fd = fd;
	fds.events = POLLIN;

	while (len) {
		fds.revents = 0;
		rc = poll(&fds, 1, timeout * 1000);
		if (rc == 0) {			/* timeout */
			return (-1);
		} else if (rc < 0) {
			if (errno == EINTR || errno == EAGAIN) {
				continue;
			} else {
				return (-1);
			}
		}

		if (!fds.revents) {
			return (-1);
		}

		bytes = read(fd, buf, len);
		if (bytes < 0) {
			if (errno == EINTR) {
				continue;
			} else {
				return (-1);
			}
		} else if (bytes == 0) {	/* eof */
			return (-1);
		}

		len -= bytes;
		buf += bytes;
	}

	return (0);
}

/*
 * read_fd() - reads data of length len from the given file descriptor fd to the
 * buffer buf, in chunks if needed. Function returns B_FALSE on failure,
 * otherwise B_TRUE. Function preserves errno, if it was set by the read(2).
 */
static boolean_t
read_fd(int fd, char *buf, size_t len)
{
	int		bytes;
#ifdef DEBUG
	size_t		len_o = len;
#endif

	while (len) {
		bytes = read(fd, buf, len);
		if (bytes < 0) {		/* err */
			if (errno == EINTR || errno == EAGAIN) {
				continue;
			} else {
				return (B_FALSE);
			}
		} else if (bytes == 0) {	/* eof */
			return (B_FALSE);
		}

		len -= bytes;
		buf += bytes;
	}

	DPRINT((dfile, "read_fd: Read %d bytes.\n", len_o - len));
	return (B_TRUE);
}

/*
 * write_fd() - writes buf of length len to the opened file descriptor fd, in
 * chunks if needed. The data from the pipe are processed in the receiving
 * thread. Function returns B_FALSE on failure, otherwise B_TRUE. Function
 * preserves errno, if it was set by the write(2).
 */
static boolean_t
write_fd(int fd, char *buf, size_t len)
{
	int		bytes;
#ifdef DEBUG
	size_t		len_o = len;
#endif

	while (len) {
		bytes = write(fd, buf, len);
		if (bytes == -1) {		/* err */
			if (errno == EINTR || errno == EAGAIN) {
				continue;
			} else {
				return (B_FALSE);
			}
		}

		len -= bytes;
		buf += bytes;
	}

	DPRINT((dfile, "write_fd: Wrote %d bytes.\n", len_o - len));
	return (B_TRUE);
}

/*
 * Plug-in entry point
 */

/*
 * send_record() - send an audit record to a host opening a connection,
 * negotiate version and establish context if necessary.
 */
send_record_rc_t
send_record(struct hostlist_s *hostlptr, const char *input, size_t in_len,
    uint64_t sequence, close_rsn_t *err_rsn)
{
	gss_buffer_desc		in_buf, out_buf;
	OM_uint32		maj_stat, min_stat;
	int			conf_state;
	int			rc;
	transq_node_t		*node_ptr;
	uint64_t		seq_n;	/* sequence in the network byte order */
	boolean_t		init_sock_poll = B_FALSE;

	/*
	 * We need to grab the reset_lock here, to prevent eventual
	 * unsynchronized cleanup calls within the reset routine (reset caused
	 * by the receiving thread) and the initialization calls in the
	 * send_record() code path.
	 */
	(void) pthread_mutex_lock(&reset_lock);

	/*
	 * Check whether the socket was closed by the recv thread prior to call
	 * send_record() and behave accordingly to the reason of the closure.
	 */
	if (recv_closure_rsn != RSN_UNDEFINED) {
		*err_rsn = recv_closure_rsn;
		if (recv_closure_rsn == RSN_GSS_CTX_EXP) {
			rc = SEND_RECORD_RETRY;
		} else {
			rc = SEND_RECORD_NEXT;
		}
		recv_closure_rsn = RSN_UNDEFINED;
		(void) pthread_mutex_unlock(&reset_lock);
		return (rc);
	}

	/*
	 * Send request to other then previously used host.
	 */
	if (current_host != hostlptr->host) {
		DPRINT((dfile, "Set new host: %s\n", hostlptr->host->h_name));
		if (sockfd != -1) {
			(void) pthread_mutex_unlock(&reset_lock);
			reset_transport(DO_CLOSE, DO_SYNC);
			return (SEND_RECORD_RETRY);
		}
		current_host = (struct hostent *)hostlptr->host;
		current_mech_oid = &hostlptr->mech;
		current_port = hostlptr->port;
	}

	/* initiate the receiving thread */
	(void) pthread_once(&recv_once_control, init_recv_record);

	/* create and connect() socket, negotiate the protocol version */
	if (sockfd == -1) {
		/* socket operations */
		DPRINT((dfile, "Socket creation and connect\n"));
		if (!sock_prepare(&sockfd, current_host, err_rsn)) {
			/* we believe the err_rsn set by sock_prepare() */
			(void) pthread_mutex_unlock(&reset_lock);
			return (SEND_RECORD_NEXT);
		}

		/* protocol version negotiation */
		DPRINT((dfile, "Protocol version negotiation\n"));
		if (prot_ver_negotiate() != 0) {
			DPRINT((dfile,
			    "Protocol version negotiation failed\n"));
			(void) pthread_mutex_unlock(&reset_lock);
			reset_transport(DO_CLOSE, DO_SYNC);
			*err_rsn = RSN_PROTOCOL_NEGOTIATE;
			return (SEND_RECORD_NEXT);
		}

		/* let the socket be initiated for poll() */
		init_sock_poll = B_TRUE;
	}

	if (!gss_ctx_initialized) {
		DPRINT((dfile, "Establishing context..\n"));
		if (establish_context() != 0) {
			(void) pthread_mutex_unlock(&reset_lock);
			reset_transport(DO_CLOSE, DO_SYNC);
			*err_rsn = RSN_GSS_CTX_ESTABLISH;
			return (SEND_RECORD_NEXT);
		}
		gss_ctx_initialized = B_TRUE;
	}

	/* let the recv thread poll() on the sockfd */
	if (init_sock_poll) {
		init_sock_poll = B_FALSE;
		if (!init_poll(sockfd)) {
			*err_rsn = RSN_INIT_POLL;
			(void) pthread_mutex_unlock(&reset_lock);
			return (SEND_RECORD_RETRY);
		}
	}

	(void) pthread_mutex_unlock(&reset_lock);

	/* if not empty, retransmit contents of the transmission queue */
	if (flush_transq) {
		DPRINT((dfile, "Retransmitting remaining (%ld) tokens from "
		    "the transmission queue\n", transq_hdr.count));
		if ((rc = transq_retransmit()) == 2) { /* gss context exp */
			reset_transport(DO_CLOSE, DO_SYNC);
			*err_rsn = RSN_GSS_CTX_EXP;
			return (SEND_RECORD_RETRY);
		} else if (rc == 1) {
			reset_transport(DO_CLOSE, DO_SYNC);
			*err_rsn = RSN_OTHER_ERR;
			return (SEND_RECORD_NEXT);
		}
		flush_transq = B_FALSE;
	}

	/*
	 * Concatenate sequence number and the new record. Note, that the
	 * pointer to the chunk of memory allocated for the concatenated values
	 * is later passed to the transq_enqueu() function which stores the
	 * pointer in the transmission queue; subsequently called
	 * transq_dequeue() frees the allocated memory once the MIC is verified
	 * by the recv_record() function.
	 *
	 * If we return earlier than the transq_enqueue() is called, it's
	 * necessary to free the in_buf.value explicitly prior to return.
	 *
	 */
	in_buf.length = in_len + sizeof (sequence);
	in_buf.value = malloc(in_buf.length);
	if (in_buf.value == NULL) {
			report_err(gettext("Memory allocation failed"));
			DPRINT((dfile, "Memory allocation failed: %s\n",
			    strerror(errno)));
			reset_transport(DO_CLOSE, DO_SYNC);
			*err_rsn = RSN_MEMORY_ALLOCATE;
			return (SEND_RECORD_FAIL);
	}
	seq_n = htonll(sequence);
	(void) memcpy(in_buf.value, &seq_n, sizeof (seq_n));
	(void) memcpy((char *)in_buf.value + sizeof (seq_n), input, in_len);

	/* wrap sequence number and the new record to the per-message token */
	(void) pthread_mutex_lock(&gss_ctx_lock);
	if (gss_ctx != NULL) {
		maj_stat = gss_wrap(&min_stat, gss_ctx, 1, GSS_C_QOP_DEFAULT,
		    &in_buf, &conf_state, &out_buf);
		(void) pthread_mutex_unlock(&gss_ctx_lock);
		switch (maj_stat) {
		case GSS_S_COMPLETE:
			break;
		case GSS_S_CONTEXT_EXPIRED:
			reset_transport(DO_CLOSE, DO_SYNC);
			free(in_buf.value);
			*err_rsn = RSN_GSS_CTX_EXP;
			return (SEND_RECORD_RETRY);
		default:
			report_gss_err(gettext("gss_wrap message"), maj_stat,
			    min_stat);
			reset_transport(DO_CLOSE, DO_SYNC);
			free(in_buf.value);
			*err_rsn = RSN_OTHER_ERR;
			return (SEND_RECORD_NEXT);
		}
	} else {	/* GSS context deleted by the recv thread */
		(void) pthread_mutex_unlock(&gss_ctx_lock);
		reset_transport(DO_CLOSE, DO_SYNC);
		free(in_buf.value);
		*err_rsn = RSN_OTHER_ERR;
		return (SEND_RECORD_NEXT);
	}


	/* enqueue the to-be-sent token into transmission queue */
	(void) pthread_mutex_lock(&transq_lock);
	if (!transq_enqueue(&node_ptr, &in_buf, sequence)) {
		(void) pthread_mutex_unlock(&transq_lock);
		reset_transport(DO_CLOSE, DO_SYNC);
		free(in_buf.value);
		(void) gss_release_buffer(&min_stat, &out_buf);
		*err_rsn = RSN_OTHER_ERR;
		return (SEND_RECORD_RETRY);
	}
	DPRINT((dfile, "Token enqueued for later verification\n"));
	(void) pthread_mutex_unlock(&transq_lock);

	/* send token */
	if (send_token(&sockfd, &out_buf) < 0) {
		DPRINT((dfile, "Token sending failed\n"));
		reset_transport(DO_CLOSE, DO_SYNC);
		(void) gss_release_buffer(&min_stat, &out_buf);

		(void) pthread_mutex_lock(&transq_lock);
		transq_dequeue(node_ptr);
		(void) pthread_mutex_unlock(&transq_lock);

		*err_rsn = RSN_OTHER_ERR;
		return (SEND_RECORD_NEXT);
	}
	DPRINT((dfile, "Token sent (transq size = %ld)\n", transq_hdr.count));

	(void) gss_release_buffer(&min_stat, &out_buf);

	return (SEND_RECORD_SUCCESS);
}

/*
 * init_recv_record() - initialize the receiver thread
 */
static void
init_recv_record()
{
	DPRINT((dfile, "Initiating the recv thread\n"));
	(void) pthread_create(&recv_tid, NULL, (void *(*)(void *))recv_record,
	    (void *)NULL);

}


/*
 * recv_record() - the receiver thread routine
 */
static void
recv_record()
{
	OM_uint32		maj_stat, min_stat;
	gss_qop_t		qop_state;
	gss_buffer_desc		in_buf = GSS_C_EMPTY_BUFFER;
	gss_buffer_desc		in_buf_mic = GSS_C_EMPTY_BUFFER;
	transq_node_t		*cur_node;
	uint64_t		r_seq_num;	/* received sequence number */
	boolean_t		token_verified;
	boolean_t		break_flag;
	struct pollfd		fds[2];
	int			fds_cnt;
	struct pollfd		*pipe_fd = &fds[0];
	struct pollfd		*recv_fd = &fds[1];
	uint32_t		len;
	int			rc;
	pipe_msg_t		np_data;

	DPRINT((dfile, "Receiver thread initiated\n"));

	/*
	 * Fill in the information in the vector of file descriptors passed
	 * later on to the poll() function. In the initial state, there is only
	 * one struct pollfd in the vector which contains file descriptor of the
	 * notification pipe - notify_pipe[1]. There might be up to two file
	 * descriptors (struct pollfd) in the vector - notify_pipe[1] which
	 * resides in the vector during the entire life of the receiving thread,
	 * and the own file descriptor from which we read data sent by the
	 * remote server application.
	 */
	pipe_fd->fd = notify_pipe[1];
	pipe_fd->events = POLLIN;
	recv_fd->fd = -1;
	recv_fd->events = POLLIN;
	fds_cnt = 1;

	/*
	 * In the endless loop, try to grab some data from the socket or
	 * notify_pipe[1].
	 */
	for (;;) {

		pipe_fd->revents = 0;
		recv_fd->revents = 0;
		recv_closure_rsn = RSN_UNDEFINED;

		/* block on poll, thus rc != 0 */
		rc = poll(fds, fds_cnt, -1);
		if (rc == -1) {
			if (errno == EAGAIN || errno == EINTR) {
				/* silently continue on EAGAIN || EINTR */
				continue;
			} else {
				/* log the debug message in any other case */
				DPRINT((dfile, "poll() failed: %s\n",
				    strerror(errno)));
				report_err(gettext("poll() failed.\n"));
				continue;
			}
		}

		/*
		 * Receive a message from the notification pipe. Information
		 * from the notification pipe takes precedence over the received
		 * data from the remote server application.
		 *
		 * Notification pipe message format - message accepted
		 * from the notify pipe comprises of two parts (int ||
		 * boolean_t), where if the first part (sizeof (int)) equals
		 * NP_CLOSE, then the second part (sizeof (boolean_t)) signals
		 * the necessity of broadcasting (DO_SYNC/DO_NOT_SYNC) the end
		 * of the reset routine.
		 */
		if (pipe_fd->revents & POLLIN) {
			DPRINT((dfile, "An event on notify pipe detected\n"));
			if (!read_fd(pipe_fd->fd, (char *)&np_data,
			    sizeof (np_data))) {
				DPRINT((dfile, "Reading notify pipe failed: "
				    "%s\n", strerror(errno)));
				report_err(gettext("Reading notify pipe "
				    "failed"));
			} else {
				switch (np_data.sock_num) {
				case NP_EXIT:	/* exit receiving thread */
					do_cleanup(&fds_cnt, recv_fd,
					    np_data.sync);
					pthread_exit((void *)NULL);
					break;
				case NP_CLOSE:	/* close and remove recv_fd */
					do_reset(&fds_cnt, recv_fd,
					    np_data.sync);
					continue;
				default:	/* add rc_pipe to the fds */
					recv_fd->fd = np_data.sock_num;
					fds_cnt = 2;
					continue;
				}
			}
		}
		/* Receive a token from the remote server application */
		if (recv_fd->revents & POLLIN) {
			DPRINT((dfile, "An event on fd detected\n"));
			if (!read_fd(recv_fd->fd, (char *)&len, sizeof (len))) {
				DPRINT((dfile, "Token length recv failed\n"));
				recv_closure_rsn = RSN_TOK_RECV_FAILED;
				reset_transport(DO_CLOSE, DO_NOT_SYNC);
				continue;
			}
			len = ntohl(len);

			/* simple DOS prevention mechanism */
			if (len > MAX_TOK_LEN) {
				report_err(gettext("Indicated invalid token "
				    "length"));
				DPRINT((dfile, "Indicated token length > %dB\n",
				    MAX_TOK_LEN));
				recv_closure_rsn = RSN_TOK_TOO_BIG;
				reset_transport(DO_CLOSE, DO_NOT_SYNC);
				continue;
			}

			in_buf.value = (char *)malloc(len);
			if (in_buf.value == NULL) {
				report_err(gettext("Memory allocation failed"));
				DPRINT((dfile, "Memory allocation failed: %s\n",
				    strerror(errno)));
				recv_closure_rsn = RSN_MEMORY_ALLOCATE;
				reset_transport(DO_CLOSE, DO_NOT_SYNC);
				continue;
			}
			if (!read_fd(recv_fd->fd, (char *)in_buf.value, len)) {
				DPRINT((dfile, "Token value recv failed\n"));
				free(in_buf.value);
				recv_closure_rsn = RSN_TOK_RECV_FAILED;
				reset_transport(DO_CLOSE, DO_NOT_SYNC);
				continue;
			}

			in_buf.length = len;
		}

		/*
		 * Extract the sequence number and the MIC from
		 * the per-message token
		 */
		(void) memcpy(&r_seq_num, in_buf.value, sizeof (r_seq_num));
		r_seq_num = ntohll(r_seq_num);
		in_buf_mic.length = in_buf.length - sizeof (r_seq_num);
		in_buf_mic.value = (char *)in_buf.value + sizeof (r_seq_num);

		/*
		 * seq_num/r_seq_num - the sequence number does not need to
		 * be unique in the transmission queue. Any token in the
		 * transmission queue with the same seq_num as the acknowledge
		 * token received from the server is tested. This is due to the
		 * fact that the plugin cannot influence (in the current
		 * implementation) sequence numbers generated by the kernel (we
		 * are reusing record sequence numbers as a transmission queue
		 * sequence numbers). The probability of having two or more
		 * tokens in the transmission queue is low and at the same time
		 * the performance gain due to using sequence numbers is quite
		 * high.
		 *
		 * In case a harder condition with regard to duplicate sequence
		 * numbers in the transmission queue will be desired over time,
		 * the break_flag behavior used below should be
		 * removed/changed_accordingly.
		 */
		break_flag = B_FALSE;
		token_verified = B_FALSE;
		(void) pthread_mutex_lock(&transq_lock);
		cur_node = transq_hdr.head;
		while (cur_node != NULL && !break_flag) {
			if (cur_node->seq_num != r_seq_num) {
				cur_node = cur_node->next;
				continue;
			}

			(void) pthread_mutex_lock(&gss_ctx_lock);
			maj_stat = gss_verify_mic(&min_stat, gss_ctx,
			    &(cur_node->seq_token), &in_buf_mic,
			    &qop_state);
			(void) pthread_mutex_unlock(&gss_ctx_lock);

			if (!GSS_ERROR(maj_stat)) { /* the success case */
				switch (maj_stat) {
				/*
				 * All the GSS_S_OLD_TOKEN, GSS_S_UNSEQ_TOKEN,
				 * GSS_S_GAP_TOKEN are perceived as correct
				 * behavior of the server side. The plugin
				 * implementation is resistant to any of the
				 * above mention cases of returned status codes.
				 */
				/*FALLTHRU*/
				case GSS_S_OLD_TOKEN:
				case GSS_S_UNSEQ_TOKEN:
				case GSS_S_GAP_TOKEN:
				case GSS_S_COMPLETE:
					/*
					 * remove the verified record/node from
					 * the transmission queue
					 */
					transq_dequeue(cur_node);
					DPRINT((dfile, "Recv thread verified "
					    "the token (transq len = %ld)\n",
					    transq_hdr.count));

					token_verified = B_TRUE;
					break_flag = B_TRUE;
					break;

				/*
				 * Both the default case as well as
				 * GSS_S_DUPLICATE_TOKEN case should never
				 * occur. It's been left here for the sake of
				 * completeness.
				 * If any of the two cases occur, it is
				 * subsequently cought because we don't set
				 * the token_verified flag.
				 */
				/*FALLTHRU*/
				case GSS_S_DUPLICATE_TOKEN:
				default:
					break_flag = B_TRUE;
					break;
				} /* switch (maj_stat) */

			} else { 	/* the failure case */
				report_gss_err(
				    gettext("signature verification of the "
				    "received token failed"),
				    maj_stat, min_stat);

				switch (maj_stat) {
				case GSS_S_CONTEXT_EXPIRED:
					/* retransmission necessary */
					recv_closure_rsn = RSN_GSS_CTX_EXP;
					break_flag = B_TRUE;
					DPRINT((dfile, "Recv thread detected "
					    "the GSS context expiration\n"));
					break;
				case GSS_S_BAD_SIG:
					DPRINT((dfile, "Bad signature "
					    "detected (seq_num = %lld)\n",
					    cur_node->seq_num));
					cur_node = cur_node->next;
					break;
				default:
					report_gss_err(
					    gettext("signature verification"),
					    maj_stat, min_stat);
					break_flag = B_TRUE;
					break;
				}
			}

		} /* while */
		(void) pthread_mutex_unlock(&transq_lock);

		if (in_buf.value != NULL) {
			free(in_buf.value);
			in_buf.value = NULL;
			in_buf.length = 0;
		}

		if (!token_verified) {
			/*
			 * Received, but unverifiable token is perceived as
			 * the protocol flow corruption with the penalty of
			 * reinitializing the client/server connection.
			 */
			DPRINT((dfile, "received unverifiable token\n"));
			report_err(gettext("received unverifiable token\n"));
			if (recv_closure_rsn == RSN_UNDEFINED) {
				recv_closure_rsn = RSN_TOK_UNVERIFIABLE;
			}
			reset_transport(DO_CLOSE, DO_NOT_SYNC);
		}

	} /* for (;;) */


}


/*
 * init_poll() - initiates the polling in the receiving thread via sending the
 * appropriate message over the notify pipe. Message format = (int ||
 * booleant_t), where the first part (sizeof (int)) contains the
 * newly_opened/to_be_polled socket file descriptor. The contents of the second
 * part (sizeof (boolean_t)) of the message works only as a padding here and no
 * action (no recv/send thread synchronisation) is made in the receiving thread
 * based on its value.
 */
static boolean_t
init_poll(int fd)
{
	pipe_msg_t	np_data;
	int		pipe_in = notify_pipe[0];

	np_data.sock_num = fd;
	np_data.sync = B_FALSE;	/* padding only */

	if (!write_fd(pipe_in, (char *)&np_data, sizeof (np_data))) {
		DPRINT((dfile, "Cannot write to the notify pipe\n"));
		report_err(gettext("writing to the notify pipe failed"));
		return (B_FALSE);
	}

	return (B_TRUE);
}


/*
 * reset_transport() - locked by the reset_lock initiates the reset of socket,
 * GSS security context and (possibly) flags the transq for retransmission; for
 * more detailed information see do_reset(). The reset_transport() also allows
 * the synchronization - waiting for the reset to be finished.
 *
 * do_close: DO_SYNC, DO_NOT_SYNC
 * sync_on_return: DO_EXIT (DO_NOT_CLOSE), DO_CLOSE (DO_NOT_EXIT)
 *
 */
void
reset_transport(boolean_t do_close, boolean_t sync_on_return)
{
	int		pipe_in = notify_pipe[0];
	pipe_msg_t	np_data;

	/*
	 * Check if the reset routine is in progress or whether it was already
	 * executed by some other thread.
	 */
	(void) pthread_mutex_lock(&reset_lock);
	if (reset_in_progress) {
		(void) pthread_mutex_unlock(&reset_lock);
		return;
	}
	reset_in_progress = B_TRUE;

	np_data.sock_num = (do_close ? NP_CLOSE : NP_EXIT);
	np_data.sync = sync_on_return;
	(void) write_fd(pipe_in, (char *)&np_data, sizeof (np_data));

	if (sync_on_return) {
		while (reset_in_progress) {
			(void) pthread_cond_wait(&reset_cv, &reset_lock);
			DPRINT((dfile, "Wait for sync\n"));
		}
		DPRINT((dfile, "Synced\n"));
	}
	(void) pthread_mutex_unlock(&reset_lock);

}


/*
 * do_reset() - the own reseting routine called from the recv thread. If the
 * synchronization was requested, signal the finish via conditional variable.
 */
static void
do_reset(int *fds_cnt, struct pollfd *recv_fd, boolean_t do_signal)
{

	(void) pthread_mutex_lock(&reset_lock);

	/* socket */
	(void) pthread_mutex_lock(&sock_lock);
	if (sockfd == -1) {
		DPRINT((dfile, "socket already closed\n"));
		(void) pthread_mutex_unlock(&sock_lock);
		goto out;
	} else {
		(void) close(sockfd);
		sockfd = -1;
		recv_fd->fd = -1;
		(void) pthread_mutex_unlock(&sock_lock);
	}
	*fds_cnt = 1;

	/* context */
	if (gss_ctx_initialized) {
		delete_context();
	}
	gss_ctx_initialized = B_FALSE;
	gss_ctx = NULL;

	/* mark transq to be flushed */
	(void) pthread_mutex_lock(&transq_lock);
	if (transq_hdr.count > 0) {
		flush_transq = B_TRUE;
	}
	(void) pthread_mutex_unlock(&transq_lock);

out:
	reset_in_progress = B_FALSE;
	if (do_signal) {
		(void) pthread_cond_broadcast(&reset_cv);
	}

	(void) pthread_mutex_unlock(&reset_lock);
}

/*
 * do_cleanup() - removes all the preallocated space by the plugin; prepares the
 * plugin/application to be gracefully finished. Even thought the function
 * allows execution without signalling the successful finish, it's recommended
 * to use it (we usually want to wait for cleanup before exiting).
 */
static void
do_cleanup(int *fds_cnt, struct pollfd *recv_fd, boolean_t do_signal)
{

	(void) pthread_mutex_lock(&reset_lock);

	/*
	 * socket
	 * note: keeping locking for safety, thought it shouldn't be necessary
	 * in current implementation - we get here only in case the sending code
	 * path calls auditd_plugin_close() (thus no socket manipulation) and
	 * the recv thread is doing the own socket closure.
	 */
	(void) pthread_mutex_lock(&sock_lock);
	if (sockfd != -1) {
		DPRINT((dfile, "Closing socket: %d\n", sockfd));
		(void) close(sockfd);
		sockfd = -1;
		recv_fd->fd = -1;
	}
	*fds_cnt = 1;
	(void) pthread_mutex_unlock(&sock_lock);

	/* context */
	if (gss_ctx_initialized) {
		DPRINT((dfile, "Deleting context: "));
		delete_context();
	}
	gss_ctx_initialized = B_FALSE;
	gss_ctx = NULL;

	/* transmission queue */
	(void) pthread_mutex_lock(&transq_lock);
	if (transq_hdr.count > 0) {
		DPRINT((dfile, "Deallocating the transmission queue "
		    "(len = %ld)\n", transq_hdr.count));
		while (transq_hdr.count > 0) {
			transq_dequeue(transq_hdr.head);
		}
	}
	(void) pthread_mutex_unlock(&transq_lock);

	/* notification pipe */
	if (notify_pipe_ready) {
		(void) close(notify_pipe[0]);
		(void) close(notify_pipe[1]);
		notify_pipe_ready = B_FALSE;
	}

	reset_in_progress = B_FALSE;
	if (do_signal) {
		(void) pthread_cond_broadcast(&reset_cv);
	}
	(void) pthread_mutex_unlock(&reset_lock);
}


/*
 * transq_dequeue() - dequeues given node pointed by the node_ptr from the
 * transmission queue. Transmission queue should be locked prior to use of this
 * function.
 */
static void
transq_dequeue(transq_node_t *node_ptr)
{

	if (node_ptr == NULL) {
		DPRINT((dfile, "transq_dequeue(): called with NULL pointer\n"));
		return;
	}

	free(node_ptr->seq_token.value);

	if (node_ptr->prev != NULL) {
		node_ptr->prev->next = node_ptr->next;
	}
	if (node_ptr->next != NULL) {
		node_ptr->next->prev = node_ptr->prev;
	}


	/* update the transq_hdr */
	if (node_ptr->next == NULL) {
		transq_hdr.end = node_ptr->prev;
	}
	if (node_ptr->prev == NULL) {
		transq_hdr.head = node_ptr->next;
	}

	transq_hdr.count--;

	free(node_ptr);
}


/*
 * transq_enqueue() - creates new node in (at the end of) the transmission
 * queue. in_ptoken_ptr is a pointer to the plain token in a form of
 * gss_buffer_desc. Function returns 0 on success and updates the *node_ptr to
 * point to a newly added transmission queue node. In case of any failure
 * function returns 1 and sets the *node_ptr to NULL.
 * Transmission queue should be locked prior to use of this function.
 */
static boolean_t
transq_enqueue(transq_node_t **node_ptr, gss_buffer_t in_seqtoken_ptr,
    uint64_t sequence)
{

	*node_ptr = calloc(1, sizeof (transq_node_t));
	if (*node_ptr == NULL) {
		report_err(gettext("Memory allocation failed"));
		DPRINT((dfile, "Memory allocation failed: %s\n",
		    strerror(errno)));
		goto errout;
	}

	/* value of the seq_token.value = (sequence number || plain token) */
	(*node_ptr)->seq_num = sequence;
	(*node_ptr)->seq_token.length = in_seqtoken_ptr->length;
	(*node_ptr)->seq_token.value = in_seqtoken_ptr->value;

	/* update the transq_hdr */
	if (transq_hdr.head == NULL) {
		transq_hdr.head = *node_ptr;
	}
	if (transq_hdr.end != NULL) {
		(transq_hdr.end)->next = *node_ptr;
		(*node_ptr)->prev = transq_hdr.end;
	}
	transq_hdr.end = *node_ptr;

	transq_hdr.count++;

	return (B_TRUE);

errout:
	if (*node_ptr != NULL) {
		if ((*node_ptr)->seq_token.value != NULL) {
			free((*node_ptr)->seq_token.value);
		}
		free(*node_ptr);
		*node_ptr = NULL;
	}
	return (B_FALSE);
}


/*
 * transq_retransmit() - traverse the transmission queue and try to, 1 by 1,
 * re-wrap the tokens with the recent context information and retransmit the
 * tokens from the transmission queue.
 * Function returns 2 on GSS context expiration, 1 on any other error, 0 on
 * successfully resent transmission queue.
 */
static int
transq_retransmit()
{

	OM_uint32	maj_stat, min_stat;
	transq_node_t	*cur_node = transq_hdr.head;
	gss_buffer_desc	out_buf;
	int		conf_state;

	DPRINT((dfile, "Retransmission of the remainder in the transqueue\n"));

	while (cur_node != NULL) {

		(void) pthread_mutex_lock(&transq_lock);
		(void) pthread_mutex_lock(&gss_ctx_lock);
		maj_stat = gss_wrap(&min_stat, gss_ctx, 1, GSS_C_QOP_DEFAULT,
		    &(cur_node->seq_token), &conf_state, &out_buf);
		(void) pthread_mutex_unlock(&gss_ctx_lock);

		switch (maj_stat) {
		case GSS_S_COMPLETE:
			break;
		case GSS_S_CONTEXT_EXPIRED:
			DPRINT((dfile, "Context expired.\n"));
			report_gss_err(gettext("gss_wrap message"), maj_stat,
			    min_stat);
			(void) pthread_mutex_unlock(&transq_lock);
			return (2);
		default:
			report_gss_err(gettext("gss_wrap message"), maj_stat,
			    min_stat);
			(void) pthread_mutex_unlock(&transq_lock);
			return (1);
		}

		DPRINT((dfile, "Sending transmission queue token (seq=%lld, "
		    "size=%d, transq len=%ld)\n", cur_node->seq_num,
		    out_buf.length, transq_hdr.count));
		if (send_token(&sockfd, &out_buf) < 0) {
			(void) gss_release_buffer(&min_stat, &out_buf);
			(void) pthread_mutex_unlock(&transq_lock);
			return (1);
		}
		(void) gss_release_buffer(&min_stat, &out_buf);

		cur_node = cur_node->next;
		(void) pthread_mutex_unlock(&transq_lock);

	} /* while */

	return (0);
}