1 files changed, 29 insertions, 87 deletions

diff --git a/src/libcharon/plugins/duplicheck/duplicheck_notify.c b/src/libcharon/plugins/duplicheck/duplicheck_notify.c
index cd5d4970b..e3a4e17b7 100644
--- a/src/libcharon/plugins/duplicheck/duplicheck_notify.c
+++ b/src/libcharon/plugins/duplicheck/duplicheck_notify.c
@@ -14,6 +14,7 @@
  */
 
 #include "duplicheck_notify.h"
+#include "duplicheck_msg.h"
 
 #include <sys/types.h>
 #include <sys/stat.h>
@@ -28,7 +29,6 @@
 #include <collections/linked_list.h>
 #include <processing/jobs/callback_job.h>
 
-#define DUPLICHECK_SOCKET IPSEC_PIDDIR "/charon.dck"
 
 typedef struct private_duplicheck_notify_t private_duplicheck_notify_t;
 
@@ -48,108 +48,53 @@ struct private_duplicheck_notify_t {
 	mutex_t *mutex;
 
 	/**
-	 * List of connected sockets
+	 * List of connected clients, as stream_t
 	 */
 	linked_list_t *connected;
 
 	/**
-	 * Socket dispatching connections
+	 * stream service accepting connections
 	 */
-	int socket;
+	stream_service_t *service;
 };
 
 /**
- * Open duplicheck unix socket
- */
-static bool open_socket(private_duplicheck_notify_t *this)
-{
-	struct sockaddr_un addr;
-	mode_t old;
-
-	addr.sun_family = AF_UNIX;
-	strcpy(addr.sun_path, DUPLICHECK_SOCKET);
-
-	this->socket = socket(AF_UNIX, SOCK_SEQPACKET, 0);
-	if (this->socket == -1)
-	{
-		DBG1(DBG_CFG, "creating duplicheck 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 duplicheck socket failed: %s", strerror(errno));
-		close(this->socket);
-		return FALSE;
-	}
-	umask(old);
-	if (chown(addr.sun_path, charon->caps->get_uid(charon->caps),
-			  charon->caps->get_gid(charon->caps)) != 0)
-	{
-		DBG1(DBG_CFG, "changing duplicheck socket permissions failed: %s",
-			 strerror(errno));
-	}
-	if (listen(this->socket, 3) < 0)
-	{
-		DBG1(DBG_CFG, "listening on duplicheck socket failed: %s",
-			 strerror(errno));
-		close(this->socket);
-		unlink(addr.sun_path);
-		return FALSE;
-	}
-	return TRUE;
-}
-
-/**
  * Accept duplicheck notification connections
  */
-static job_requeue_t receive(private_duplicheck_notify_t *this)
+static bool on_accept(private_duplicheck_notify_t *this, stream_t *stream)
 {
-	struct sockaddr_un addr;
-	int len = sizeof(addr);
-	uintptr_t fd;
-	bool oldstate;
-
-	oldstate = thread_cancelability(TRUE);
-	fd = accept(this->socket, (struct sockaddr*)&addr, &len);
-	thread_cancelability(oldstate);
+	this->mutex->lock(this->mutex);
+	this->connected->insert_last(this->connected, stream);
+	this->mutex->unlock(this->mutex);
 
-	if (fd != -1)
-	{
-		this->mutex->lock(this->mutex);
-		this->connected->insert_last(this->connected, (void*)fd);
-		this->mutex->unlock(this->mutex);
-	 }
-	 else
-	 {
-		 DBG1(DBG_CFG, "accepting duplicheck connection failed: %s",
-			  strerror(errno));
-	 }
-	 return JOB_REQUEUE_FAIR;
+	return TRUE;
 }
 
 METHOD(duplicheck_notify_t, send_, void,
 	private_duplicheck_notify_t *this, identification_t *id)
 {
-	char buf[128];
 	enumerator_t *enumerator;
-	uintptr_t fd;
+	stream_t *stream;
+	u_int16_t nlen;
+	char buf[512];
 	int len;
 
 	len = snprintf(buf, sizeof(buf), "%Y", id);
 	if (len > 0 && len < sizeof(buf))
 	{
+		nlen = htons(len);
+
 		this->mutex->lock(this->mutex);
 		enumerator = this->connected->create_enumerator(this->connected);
-		while (enumerator->enumerate(enumerator, &fd))
+		while (enumerator->enumerate(enumerator, &stream))
 		{
-			if (send(fd, &buf, len + 1, 0) != len + 1)
+			if (!stream->write_all(stream, &nlen, sizeof(nlen)) ||
+				!stream->write_all(stream, buf, len))
 			{
 				DBG1(DBG_CFG, "sending duplicheck notify failed: %s",
 					 strerror(errno));
 				this->connected->remove_at(this->connected, enumerator);
-				close(fd);
+				stream->destroy(stream);
 			}
 		}
 		enumerator->destroy(enumerator);
@@ -160,16 +105,8 @@ METHOD(duplicheck_notify_t, send_, void,
 METHOD(duplicheck_notify_t, destroy, void,
 	private_duplicheck_notify_t *this)
 {
-	enumerator_t *enumerator;
-	uintptr_t fd;
-
-	enumerator = this->connected->create_enumerator(this->connected);
-	while (enumerator->enumerate(enumerator, &fd))
-	{
-		close(fd);
-	}
-	enumerator->destroy(enumerator);
-	this->connected->destroy(this->connected);
+	DESTROY_IF(this->service);
+	this->connected->destroy_offset(this->connected, offsetof(stream_t, destroy));
 	this->mutex->destroy(this->mutex);
 	free(this);
 }
@@ -180,6 +117,7 @@ METHOD(duplicheck_notify_t, destroy, void,
 duplicheck_notify_t *duplicheck_notify_create()
 {
 	private_duplicheck_notify_t *this;
+	char *uri;
 
 	INIT(this,
 		.public = {
@@ -190,14 +128,18 @@ duplicheck_notify_t *duplicheck_notify_create()
 		.mutex = mutex_create(MUTEX_TYPE_DEFAULT),
 	);
 
-	if (!open_socket(this))
+	uri = lib->settings->get_str(lib->settings,
+					"%s.plugins.duplicheck.socket", "unix://" DUPLICHECK_SOCKET,
+					charon->name);
+	this->service = lib->streams->create_service(lib->streams, uri, 3);
+	if (!this->service)
 	{
+		DBG1(DBG_CFG, "creating duplicheck socket failed");
 		destroy(this);
 		return NULL;
 	}
-	lib->processor->queue_job(lib->processor,
-		(job_t*)callback_job_create_with_prio((callback_job_cb_t)receive, this,
-				NULL, (callback_job_cancel_t)return_false, JOB_PRIO_CRITICAL));
+	this->service->on_accept(this->service, (stream_service_cb_t)on_accept,
+							 this, JOB_PRIO_CRITICAL, 1);
 
 	return &this->public;
 }