error-notify: use a stream service to accept client connections

As TCP does not have SOCK_SEQPACKET, we now use SOCK_STREAM for the error-notify
socket. To have network transparency, the message now uses network byte order.

src/libcharon/plugins/error_notify/error_notify.c

@@ -16,46 +16,89 @@
 #include "error_notify_msg.h"
 
 #include <stdio.h>
+#include <stdlib.h>
+#include <stddef.h>
 #include <unistd.h>
 #include <sys/stat.h>
 #include <sys/socket.h>
 #include <sys/un.h>
 #include <errno.h>
+#include <arpa/inet.h>
+
+/**
+ * Connect to the daemon, return FD
+ */
+static int make_connection()
+{
+	union {
+		struct sockaddr_un un;
+		struct sockaddr_in in;
+		struct sockaddr sa;
+	} addr;
+	int fd, len;
+
+	if (getenv("TCP_PORT"))
+	{
+		addr.in.sin_family = AF_INET;
+		addr.in.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+		addr.in.sin_port = htons(atoi(getenv("TCP_PORT")));
+		len = sizeof(addr.in);
+	}
+	else
+	{
+		addr.un.sun_family = AF_UNIX;
+		strcpy(addr.un.sun_path, ERROR_NOTIFY_SOCKET);
+
+		len = offsetof(struct sockaddr_un, sun_path) + strlen(addr.un.sun_path);
+	}
+	fd = socket(addr.sa.sa_family, SOCK_STREAM, 0);
+	if (fd < 0)
+	{
+		fprintf(stderr, "opening socket failed: %s\n", strerror(errno));
+		return -1;
+	}
+	if (connect(fd, &addr.sa, len) < 0)
+	{
+		fprintf(stderr, "connecting failed: %s\n", strerror(errno));
+		close(fd);
+		return -1;
+	}
+	return fd;
+}
 
 /**
  * Example of a simple notification listener
  */
 int main(int argc, char *argv[])
 {
-	struct sockaddr_un addr;
 	error_notify_msg_t msg;
-	int s;
+	int s, len, total;
+	void *pos;
 
-	addr.sun_family = AF_UNIX;
+	s = make_connection();
-	strcpy(addr.sun_path, ERROR_NOTIFY_SOCKET);
-
-	s = socket(AF_UNIX, SOCK_SEQPACKET, 0);
 	if (s < 0)
 	{
-		fprintf(stderr, "opening socket failed: %s\n", strerror(errno));
-		return 1;
-	}
-	if (connect(s, (struct sockaddr *)&addr, sizeof(addr)) < 0)
-	{
-		fprintf(stderr, "connect failed: %s\n", strerror(errno));
-		close(s);
 		return 1;
 	}
 	while (1)
 	{
-		if (read(s, &msg, sizeof(msg)) != sizeof(msg))
+		total = 0;
+		pos = &msg;
+
+		while (total < sizeof(msg))
 		{
-			fprintf(stderr, "read failed: %s\n", strerror(errno));
+			len = read(s, pos, sizeof(msg) - total);
-			close(s);
+			if (len < 0)
-			return 1;
+			{
+				fprintf(stderr, "read failed: %s\n", strerror(errno));
+				close(s);
+				return 1;
+			}
+			total += len;
+			pos += len;
 		}
 		printf("%d %s %s %s %s\n",
-			   msg.type, msg.name, msg.id, msg.ip, msg.str);
+			   ntohl(msg.type), msg.name, msg.id, msg.ip, msg.str);
 	}
 	close(s);
 	return 0;

src/libcharon/plugins/error_notify/error_notify_listener.c

@@ -56,80 +56,80 @@ METHOD(listener_t, alert, bool,
 	switch (alert)
 	{
 		case ALERT_RADIUS_NOT_RESPONDING:
-			msg.type = ERROR_NOTIFY_RADIUS_NOT_RESPONDING;
+			msg.type = htonl(ERROR_NOTIFY_RADIUS_NOT_RESPONDING);
 			snprintf(msg.str, sizeof(msg.str),
 					 "a RADIUS request message timed out");
 			break;
 		case ALERT_LOCAL_AUTH_FAILED:
-			msg.type = ERROR_NOTIFY_LOCAL_AUTH_FAILED;
+			msg.type = htonl(ERROR_NOTIFY_LOCAL_AUTH_FAILED);
 			snprintf(msg.str, sizeof(msg.str),
 					 "creating local authentication data failed");
 			break;
 		case ALERT_PEER_AUTH_FAILED:
-			msg.type = ERROR_NOTIFY_PEER_AUTH_FAILED;
+			msg.type = htonl(ERROR_NOTIFY_PEER_AUTH_FAILED);
 			snprintf(msg.str, sizeof(msg.str), "peer authentication failed");
 			break;
 		case ALERT_PARSE_ERROR_HEADER:
-			msg.type = ERROR_NOTIFY_PARSE_ERROR_HEADER;
+			msg.type = htonl(ERROR_NOTIFY_PARSE_ERROR_HEADER);
 			message = va_arg(args, message_t*);
 			snprintf(msg.str, sizeof(msg.str), "parsing IKE header from "
 					 "%#H failed", message->get_source(message));
 			break;
 		case ALERT_PARSE_ERROR_BODY:
-			msg.type = ERROR_NOTIFY_PARSE_ERROR_BODY;
+			msg.type = htonl(ERROR_NOTIFY_PARSE_ERROR_BODY);
 			message = va_arg(args, message_t*);
 			snprintf(msg.str, sizeof(msg.str), "parsing IKE message from "
 					 "%#H failed", message->get_source(message));
 			break;
 		case ALERT_RETRANSMIT_SEND_TIMEOUT:
-			msg.type = ERROR_NOTIFY_RETRANSMIT_SEND_TIMEOUT;
+			msg.type = htonl(ERROR_NOTIFY_RETRANSMIT_SEND_TIMEOUT);
 			snprintf(msg.str, sizeof(msg.str),
 					 "IKE message retransmission timed out");
 			break;
 		case ALERT_HALF_OPEN_TIMEOUT:
-			msg.type = ERROR_NOTIFY_HALF_OPEN_TIMEOUT;
+			msg.type = htonl(ERROR_NOTIFY_HALF_OPEN_TIMEOUT);
 			snprintf(msg.str, sizeof(msg.str), "IKE_SA timed out before it "
 					 "could be established");
 			break;
 		case ALERT_PROPOSAL_MISMATCH_IKE:
-			msg.type = ERROR_NOTIFY_PROPOSAL_MISMATCH_IKE;
+			msg.type = htonl(ERROR_NOTIFY_PROPOSAL_MISMATCH_IKE);
 			list = va_arg(args, linked_list_t*);
 			snprintf(msg.str, sizeof(msg.str), "the received IKE_SA poposals "
 					 "did not match: %#P", list);
 			break;
 		case ALERT_PROPOSAL_MISMATCH_CHILD:
-			msg.type = ERROR_NOTIFY_PROPOSAL_MISMATCH_CHILD;
+			msg.type = htonl(ERROR_NOTIFY_PROPOSAL_MISMATCH_CHILD);
 			list = va_arg(args, linked_list_t*);
 			snprintf(msg.str, sizeof(msg.str), "the received CHILD_SA poposals "
 					 "did not match: %#P", list);
 			break;
 		case ALERT_TS_MISMATCH:
-			msg.type = ERROR_NOTIFY_TS_MISMATCH;
+			msg.type = htonl(ERROR_NOTIFY_TS_MISMATCH);
 			list = va_arg(args, linked_list_t*);
 			list2 = va_arg(args, linked_list_t*);
 			snprintf(msg.str, sizeof(msg.str), "the received traffic selectors "
 					 "did not match: %#R=== %#R", list, list2);
 			break;
 		case ALERT_INSTALL_CHILD_SA_FAILED:
-			msg.type = ERROR_NOTIFY_INSTALL_CHILD_SA_FAILED;
+			msg.type = htonl(ERROR_NOTIFY_INSTALL_CHILD_SA_FAILED);
 			snprintf(msg.str, sizeof(msg.str), "installing IPsec SA failed");
 			break;
 		case ALERT_INSTALL_CHILD_POLICY_FAILED:
-			msg.type = ERROR_NOTIFY_INSTALL_CHILD_POLICY_FAILED;
+			msg.type = htonl(ERROR_NOTIFY_INSTALL_CHILD_POLICY_FAILED);
 			snprintf(msg.str, sizeof(msg.str), "installing IPsec policy failed");
 			break;
 		case ALERT_UNIQUE_REPLACE:
-			msg.type = ERROR_NOTIFY_UNIQUE_REPLACE;
+			msg.type = htonl(ERROR_NOTIFY_UNIQUE_REPLACE);
 			snprintf(msg.str, sizeof(msg.str),
 					 "replaced old IKE_SA due to uniqueness policy");
 			break;
 		case ALERT_UNIQUE_KEEP:
-			msg.type = ERROR_NOTIFY_UNIQUE_KEEP;
+			msg.type = htonl(ERROR_NOTIFY_UNIQUE_KEEP);
 			snprintf(msg.str, sizeof(msg.str), "keep existing in favor of "
 					 "rejected new IKE_SA due to uniqueness policy");
 			break;
 		case ALERT_VIP_FAILURE:
-			msg.type = ERROR_NOTIFY_VIP_FAILURE;
+			msg.type = htonl(ERROR_NOTIFY_VIP_FAILURE);
 			list = va_arg(args, linked_list_t*);
 			if (list->get_first(list, (void**)&host) == SUCCESS)
 			{
@@ -143,7 +143,7 @@ METHOD(listener_t, alert, bool,
 			}
 			break;
 		case ALERT_AUTHORIZATION_FAILED:
-			msg.type = ERROR_NOTIFY_AUTHORIZATION_FAILED;
+			msg.type = htonl(ERROR_NOTIFY_AUTHORIZATION_FAILED);
 			snprintf(msg.str, sizeof(msg.str), "an authorization plugin "
 					 "prevented establishment of an IKE_SA");
 			break;

src/libcharon/plugins/error_notify/error_notify_msg.h

@@ -61,6 +61,6 @@ struct error_notify_msg_t {
 	char id[128];
 	/** peer address and port, if known */
 	char ip[60];
-};
+} __attribute__((packed));
 
 #endif /** ERROR_NOTIFY_MSG_H_ @}*/

src/libcharon/plugins/error_notify/error_notify_socket.c

@@ -43,12 +43,12 @@ struct private_error_notify_socket_t {
 	error_notify_socket_t public;
 
 	/**
-	 * Unix socket file descriptor
+	 * Service accepting connections
 	 */
-	int socket;
+	stream_service_t *service;
 
 	/**
-	 * List of connected clients, as uintptr_t FD
+	 * List of connected clients, as stream_t
 	 */
 	linked_list_t *connected;
 
@@ -58,48 +58,6 @@ struct private_error_notify_socket_t {
 	mutex_t *mutex;
 };
 
-/**
- * Open error notify unix socket
- */
-static bool open_socket(private_error_notify_socket_t *this)
-{
-	struct sockaddr_un addr;
-	mode_t old;
-
-	addr.sun_family = AF_UNIX;
-	strcpy(addr.sun_path, ERROR_NOTIFY_SOCKET);
-
-	this->socket = socket(AF_UNIX, SOCK_SEQPACKET, 0);
-	if (this->socket == -1)
-	{
-		DBG1(DBG_CFG, "creating notify socket failed");
-		return FALSE;
-	}
-	unlink(addr.sun_path);
-	old = umask(~(S_IRWXU | S_IRWXG));
-	if (bind(this->socket, (struct sockaddr*)&addr, sizeof(addr)) < 0)
-	{
-		DBG1(DBG_CFG, "binding notify socket failed: %s", strerror(errno));
-		close(this->socket);
-		return FALSE;
-	}
-	umask(old);
-	if (chown(addr.sun_path, lib->caps->get_uid(lib->caps),
-			  lib->caps->get_gid(lib->caps)) != 0)
-	{
-		DBG1(DBG_CFG, "changing notify socket permissions failed: %s",
-			 strerror(errno));
-	}
-	if (listen(this->socket, 10) < 0)
-	{
-		DBG1(DBG_CFG, "listening on notify socket failed: %s", strerror(errno));
-		close(this->socket);
-		unlink(addr.sun_path);
-		return FALSE;
-	}
-	return TRUE;
-}
-
 METHOD(error_notify_socket_t, has_listeners, bool,
 	private_error_notify_socket_t *this)
 {
@@ -116,23 +74,21 @@ METHOD(error_notify_socket_t, notify, void,
 	private_error_notify_socket_t *this, error_notify_msg_t *msg)
 {
 	enumerator_t *enumerator;
-	uintptr_t fd;
+	stream_t *stream;
 
 	this->mutex->lock(this->mutex);
 	enumerator = this->connected->create_enumerator(this->connected);
-	while (enumerator->enumerate(enumerator, (void*)&fd))
+	while (enumerator->enumerate(enumerator, &stream))
 	{
-		while (send(fd, msg, sizeof(*msg), 0) <= 0)
+		if (!stream->write_all(stream, msg, sizeof(*msg)))
 		{
 			switch (errno)
 			{
-				case EINTR:
-					continue;
 				case ECONNRESET:
 				case EPIPE:
 					/* disconnect, remove this listener */
 					this->connected->remove_at(this->connected, enumerator);
-					close(fd);
+					stream->destroy(stream);
 					break;
 				default:
 					DBG1(DBG_CFG, "sending notify failed: %s", strerror(errno));
@@ -146,45 +102,23 @@ METHOD(error_notify_socket_t, notify, void,
 }
 
 /**
- * Accept client connections, dispatch
+ * Accept client connections
  */
-static job_requeue_t accept_(private_error_notify_socket_t *this)
+static bool on_accept(private_error_notify_socket_t *this, stream_t *stream)
 {
-	struct sockaddr_un addr;
+	this->mutex->lock(this->mutex);
-	int fd, len;
+	this->connected->insert_last(this->connected, stream);
-	bool oldstate;
+	this->mutex->unlock(this->mutex);
 
-	len = sizeof(addr);
+	return TRUE;
-	oldstate = thread_cancelability(TRUE);
-	fd = accept(this->socket, (struct sockaddr*)&addr, &len);
-	thread_cancelability(oldstate);
-
-	if (fd != -1)
-	{
-		this->mutex->lock(this->mutex);
-		this->connected->insert_last(this->connected, (void*)(uintptr_t)fd);
-		this->mutex->unlock(this->mutex);
-	}
-	else
-	{
-		DBG1(DBG_CFG, "accepting notify connection failed: %s",
-			 strerror(errno));
-	}
-	return JOB_REQUEUE_DIRECT;
 }
 
 METHOD(error_notify_socket_t, destroy, void,
 	private_error_notify_socket_t *this)
 {
-	uintptr_t fd;
+	DESTROY_IF(this->service);
-
+	this->connected->destroy_offset(this->connected, offsetof(stream_t, destroy));
-	while (this->connected->remove_last(this->connected, (void*)&fd) == SUCCESS)
-	{
-		close(fd);
-	}
-	this->connected->destroy(this->connected);
 	this->mutex->destroy(this->mutex);
-	close(this->socket);
 	free(this);
 }
 
@@ -194,6 +128,7 @@ METHOD(error_notify_socket_t, destroy, void,
 error_notify_socket_t *error_notify_socket_create()
 {
 	private_error_notify_socket_t *this;
+	char *uri;
 
 	INIT(this,
 		.public = {
@@ -205,15 +140,18 @@ error_notify_socket_t *error_notify_socket_create()
 		.mutex = mutex_create(MUTEX_TYPE_DEFAULT),
 	);
 
-	if (!open_socket(this))
+	uri = lib->settings->get_str(lib->settings,
+				"%s.plugins.error-notify.socket", "unix://" ERROR_NOTIFY_SOCKET,
+				charon->name);
+	this->service = lib->streams->create_service(lib->streams, uri, 10);
+	if (!this->service)
 	{
-		free(this);
+		DBG1(DBG_CFG, "creating duplicheck socket failed");
+		destroy(this);
 		return NULL;
 	}
-
+	this->service->on_accept(this->service, (stream_service_cb_t)on_accept,
-	lib->processor->queue_job(lib->processor,
+							 this, JOB_PRIO_CRITICAL, 1);
-		(job_t*)callback_job_create_with_prio((callback_job_cb_t)accept_, this,
-				NULL, (callback_job_cancel_t)return_false, JOB_PRIO_CRITICAL));
 
 	return &this->public;
 }