Adding upstream version 2:4.17.12+dfsg.upstream/2%4.17.12+dfsgupstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

10 files changed, 3527 insertions, 0 deletions

diff --git a/source4/lib/messaging/irpc.h b/source4/lib/messaging/irpc.h
new file mode 100644
index 0000000..ebf30af
--- /dev/null
+++ b/source4/lib/messaging/irpc.h
@@ -0,0 +1,86 @@
+/* 
+   Unix SMB/CIFS implementation.
+
+   Samba internal rpc code - header
+
+   Copyright (C) Andrew Tridgell 2005
+   
+   This program is free software; you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation; either version 3 of the License, or
+   (at your option) any later version.
+   
+   This program 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 General Public License for more details.
+   
+   You should have received a copy of the GNU General Public License
+   along with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#ifndef IRPC_H
+#define IRPC_H
+
+#include "lib/messaging/messaging.h"
+#include "librpc/gen_ndr/irpc.h"
+
+/*
+  an incoming irpc message
+*/
+struct irpc_message {
+	struct server_id from;
+	void *private_data;
+	struct irpc_header header;
+	struct ndr_pull *ndr;
+	bool defer_reply;
+	bool no_reply;
+	struct imessaging_context *msg_ctx;
+	struct irpc_list *irpc;
+	void *data;
+};
+
+/* don't allow calls to take too long */
+#define IRPC_CALL_TIMEOUT	10
+/* wait for the calls as long as it takes */
+#define IRPC_CALL_TIMEOUT_INF	0
+
+
+/* the server function type */
+typedef NTSTATUS (*irpc_function_t)(struct irpc_message *, void *r);
+
+/* register a server function with the irpc messaging system */
+#define IRPC_REGISTER(msg_ctx, pipename, funcname, function, private_data) \
+   irpc_register(msg_ctx, &ndr_table_ ## pipename, \
+                          NDR_ ## funcname, \
+			  (irpc_function_t)function, private_data)
+
+struct ndr_interface_table;
+
+NTSTATUS irpc_register(struct imessaging_context *msg_ctx,
+		       const struct ndr_interface_table *table, 
+		       int call, irpc_function_t fn, void *private_data);
+
+struct dcerpc_binding_handle *irpc_binding_handle(TALLOC_CTX *mem_ctx,
+						  struct imessaging_context *msg_ctx,
+						  struct server_id server_id,
+						  const struct ndr_interface_table *table);
+struct dcerpc_binding_handle *irpc_binding_handle_by_name(TALLOC_CTX *mem_ctx,
+							  struct imessaging_context *msg_ctx,
+							  const char *dest_task,
+							  const struct ndr_interface_table *table);
+void irpc_binding_handle_add_security_token(struct dcerpc_binding_handle *h,
+					    struct security_token *token);
+
+NTSTATUS irpc_add_name(struct imessaging_context *msg_ctx, const char *name);
+NTSTATUS irpc_servers_byname(struct imessaging_context *msg_ctx,
+			     TALLOC_CTX *mem_ctx, const char *name,
+			     unsigned *num_servers,
+			     struct server_id **servers);
+struct irpc_name_records *irpc_all_servers(struct imessaging_context *msg_ctx,
+					   TALLOC_CTX *mem_ctx);
+void irpc_remove_name(struct imessaging_context *msg_ctx, const char *name);
+NTSTATUS irpc_send_reply(struct irpc_message *m, NTSTATUS status);
+
+#endif
+
diff --git a/source4/lib/messaging/messaging.c b/source4/lib/messaging/messaging.c
new file mode 100644
index 0000000..67b0366
--- /dev/null
+++ b/source4/lib/messaging/messaging.c
@@ -0,0 +1,1485 @@
+/*
+   Unix SMB/CIFS implementation.
+
+   Samba internal messaging functions
+
+   Copyright (C) Andrew Tridgell 2004
+
+   This program is free software; you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation; either version 3 of the License, or
+   (at your option) any later version.
+
+   This program 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 General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#include "includes.h"
+#include "lib/events/events.h"
+#include "lib/util/server_id.h"
+#include "system/filesys.h"
+#include "messaging/messaging.h"
+#include "messaging/messaging_internal.h"
+#include "../lib/util/dlinklist.h"
+#include "lib/socket/socket.h"
+#include "librpc/gen_ndr/ndr_irpc.h"
+#include "lib/messaging/irpc.h"
+#include "../lib/util/unix_privs.h"
+#include "librpc/rpc/dcerpc.h"
+#include "cluster/cluster.h"
+#include "../lib/util/tevent_ntstatus.h"
+#include "lib/param/param.h"
+#include "lib/util/server_id_db.h"
+#include "lib/util/talloc_report_printf.h"
+#include "lib/messaging/messages_dgm.h"
+#include "lib/messaging/messages_dgm_ref.h"
+#include "../source3/lib/messages_util.h"
+#include <tdb.h>
+
+/* change the message version with any incompatible changes in the protocol */
+#define IMESSAGING_VERSION 1
+
+/*
+  a pending irpc call
+*/
+struct irpc_request {
+	struct irpc_request *prev, *next;
+	struct imessaging_context *msg_ctx;
+	int callid;
+	struct {
+		void (*handler)(struct irpc_request *irpc, struct irpc_message *m);
+		void *private_data;
+	} incoming;
+};
+
+/* we have a linked list of dispatch handlers for each msg_type that
+   this messaging server can deal with */
+struct dispatch_fn {
+	struct dispatch_fn *next, *prev;
+	uint32_t msg_type;
+	void *private_data;
+	msg_callback_t fn;
+};
+
+/* an individual message */
+
+static void irpc_handler(struct imessaging_context *,
+			 void *,
+			 uint32_t,
+			 struct server_id,
+			 size_t,
+			 int *,
+			 DATA_BLOB *);
+
+
+/*
+ A useful function for testing the message system.
+*/
+static void ping_message(struct imessaging_context *msg,
+			 void *private_data,
+			 uint32_t msg_type,
+			 struct server_id src,
+			 size_t num_fds,
+			 int *fds,
+			 DATA_BLOB *data)
+{
+	struct server_id_buf idbuf;
+
+	if (num_fds != 0) {
+		DBG_WARNING("Received %zu fds, ignoring message\n", num_fds);
+		return;
+	}
+
+	DEBUG(1,("INFO: Received PING message from server %s [%.*s]\n",
+		 server_id_str_buf(src, &idbuf), (int)data->length,
+		 data->data?(const char *)data->data:""));
+	imessaging_send(msg, src, MSG_PONG, data);
+}
+
+static void pool_message(struct imessaging_context *msg,
+			 void *private_data,
+			 uint32_t msg_type,
+			 struct server_id src,
+			 size_t num_fds,
+			 int *fds,
+			 DATA_BLOB *data)
+{
+	FILE *f = NULL;
+
+	if (num_fds != 1) {
+		DBG_WARNING("Received %zu fds, ignoring message\n", num_fds);
+		return;
+	}
+
+	f = fdopen(fds[0], "w");
+	if (f == NULL) {
+		DBG_DEBUG("fopen failed: %s\n", strerror(errno));
+		return;
+	}
+
+	talloc_full_report_printf(NULL, f);
+	fclose(f);
+}
+
+static void ringbuf_log_msg(struct imessaging_context *msg,
+			    void *private_data,
+			    uint32_t msg_type,
+			    struct server_id src,
+			    size_t num_fds,
+			    int *fds,
+			    DATA_BLOB *data)
+{
+	char *log = debug_get_ringbuf();
+	size_t logsize = debug_get_ringbuf_size();
+	DATA_BLOB blob;
+
+	if (num_fds != 0) {
+		DBG_WARNING("Received %zu fds, ignoring message\n", num_fds);
+		return;
+	}
+
+	if (log == NULL) {
+		log = discard_const_p(char, "*disabled*\n");
+		logsize = strlen(log) + 1;
+	}
+
+	blob.data = (uint8_t *)log;
+	blob.length = logsize;
+
+	imessaging_send(msg, src, MSG_RINGBUF_LOG, &blob);
+}
+
+/****************************************************************************
+ Receive a "set debug level" message.
+****************************************************************************/
+
+static void debug_imessage(struct imessaging_context *msg_ctx,
+			   void *private_data,
+			   uint32_t msg_type,
+			   struct server_id src,
+			   size_t num_fds,
+			   int *fds,
+			   DATA_BLOB *data)
+{
+	const char *params_str = (const char *)data->data;
+	struct server_id_buf src_buf;
+	struct server_id dst = imessaging_get_server_id(msg_ctx);
+	struct server_id_buf dst_buf;
+
+	if (num_fds != 0) {
+		DBG_WARNING("Received %zu fds, ignoring message\n", num_fds);
+		return;
+	}
+
+	/* Check, it's a proper string! */
+	if (params_str[(data->length)-1] != '\0') {
+		DBG_ERR("Invalid debug message from pid %s to pid %s\n",
+			server_id_str_buf(src, &src_buf),
+			server_id_str_buf(dst, &dst_buf));
+		return;
+	}
+
+	DBG_ERR("INFO: Remote set of debug to `%s' (pid %s from pid %s)\n",
+		params_str,
+		server_id_str_buf(dst, &dst_buf),
+		server_id_str_buf(src, &src_buf));
+
+	debug_parse_levels(params_str);
+}
+
+/****************************************************************************
+ Return current debug level.
+****************************************************************************/
+
+static void debuglevel_imessage(struct imessaging_context *msg_ctx,
+				void *private_data,
+				uint32_t msg_type,
+				struct server_id src,
+				size_t num_fds,
+				int *fds,
+				DATA_BLOB *data)
+{
+	char *message = debug_list_class_names_and_levels();
+	DATA_BLOB blob = data_blob_null;
+	struct server_id_buf src_buf;
+	struct server_id dst = imessaging_get_server_id(msg_ctx);
+	struct server_id_buf dst_buf;
+
+	if (num_fds != 0) {
+		DBG_WARNING("Received %zu fds, ignoring message\n", num_fds);
+		return;
+	}
+
+	DBG_DEBUG("Received REQ_DEBUGLEVEL message (pid %s from pid %s)\n",
+		  server_id_str_buf(dst, &dst_buf),
+		  server_id_str_buf(src, &src_buf));
+
+	if (message == NULL) {
+		DBG_ERR("debug_list_class_names_and_levels returned NULL\n");
+		return;
+	}
+
+	blob = data_blob_string_const_null(message);
+	imessaging_send(msg_ctx, src, MSG_DEBUGLEVEL, &blob);
+
+	TALLOC_FREE(message);
+}
+
+/*
+  return uptime of messaging server via irpc
+*/
+static NTSTATUS irpc_uptime(struct irpc_message *msg,
+			    struct irpc_uptime *r)
+{
+	struct imessaging_context *ctx = talloc_get_type(msg->private_data, struct imessaging_context);
+	*r->out.start_time = timeval_to_nttime(&ctx->start_time);
+	return NT_STATUS_OK;
+}
+
+static struct dispatch_fn *imessaging_find_dispatch(
+	struct imessaging_context *msg, uint32_t msg_type)
+{
+	/* temporary IDs use an idtree, the rest use a array of pointers */
+	if (msg_type >= MSG_TMP_BASE) {
+		return (struct dispatch_fn *)idr_find(msg->dispatch_tree,
+						      msg_type);
+	}
+	if (msg_type < msg->num_types) {
+		return msg->dispatch[msg_type];
+	}
+	return NULL;
+}
+
+/*
+  Register a dispatch function for a particular message type.
+*/
+NTSTATUS imessaging_register(struct imessaging_context *msg, void *private_data,
+			    uint32_t msg_type, msg_callback_t fn)
+{
+	struct dispatch_fn *d;
+
+	/* possibly expand dispatch array */
+	if (msg_type >= msg->num_types) {
+		struct dispatch_fn **dp;
+		uint32_t i;
+		dp = talloc_realloc(msg, msg->dispatch, struct dispatch_fn *, msg_type+1);
+		NT_STATUS_HAVE_NO_MEMORY(dp);
+		msg->dispatch = dp;
+		for (i=msg->num_types;i<=msg_type;i++) {
+			msg->dispatch[i] = NULL;
+		}
+		msg->num_types = msg_type+1;
+	}
+
+	d = talloc_zero(msg->dispatch, struct dispatch_fn);
+	NT_STATUS_HAVE_NO_MEMORY(d);
+	d->msg_type = msg_type;
+	d->private_data = private_data;
+	d->fn = fn;
+
+	DLIST_ADD(msg->dispatch[msg_type], d);
+
+	return NT_STATUS_OK;
+}
+
+/*
+  register a temporary message handler. The msg_type is allocated
+  above MSG_TMP_BASE
+*/
+NTSTATUS imessaging_register_tmp(struct imessaging_context *msg, void *private_data,
+				msg_callback_t fn, uint32_t *msg_type)
+{
+	struct dispatch_fn *d;
+	int id;
+
+	d = talloc_zero(msg->dispatch, struct dispatch_fn);
+	NT_STATUS_HAVE_NO_MEMORY(d);
+	d->private_data = private_data;
+	d->fn = fn;
+
+	id = idr_get_new_above(msg->dispatch_tree, d, MSG_TMP_BASE, UINT16_MAX);
+	if (id == -1) {
+		talloc_free(d);
+		return NT_STATUS_TOO_MANY_CONTEXT_IDS;
+	}
+
+	d->msg_type = (uint32_t)id;
+	(*msg_type) = d->msg_type;
+
+	return NT_STATUS_OK;
+}
+
+/*
+  De-register the function for a particular message type.
+*/
+void imessaging_deregister(struct imessaging_context *msg, uint32_t msg_type, void *private_data)
+{
+	struct dispatch_fn *d, *next;
+
+	if (msg_type >= msg->num_types) {
+		d = (struct dispatch_fn *)idr_find(msg->dispatch_tree,
+						   msg_type);
+		if (!d) return;
+		idr_remove(msg->dispatch_tree, msg_type);
+		talloc_free(d);
+		return;
+	}
+
+	for (d = msg->dispatch[msg_type]; d; d = next) {
+		next = d->next;
+		if (d->private_data == private_data) {
+			DLIST_REMOVE(msg->dispatch[msg_type], d);
+			talloc_free(d);
+		}
+	}
+}
+
+/*
+*/
+int imessaging_cleanup(struct imessaging_context *msg)
+{
+	if (!msg) {
+		return 0;
+	}
+	return 0;
+}
+
+static void imessaging_dgm_recv(struct tevent_context *ev,
+				const uint8_t *buf, size_t buf_len,
+				int *fds, size_t num_fds,
+				void *private_data);
+
+/* Keep a list of imessaging contexts */
+static struct imessaging_context *msg_ctxs;
+
+/*
+ * A process has terminated, clean-up any names it has registered.
+ */
+NTSTATUS imessaging_process_cleanup(
+	struct imessaging_context *msg_ctx,
+	pid_t pid)
+{
+	struct irpc_name_records *names = NULL;
+	uint32_t i = 0;
+	uint32_t j = 0;
+	TALLOC_CTX *mem_ctx = talloc_new(NULL);
+
+	if (mem_ctx == NULL) {
+		DBG_ERR("OOM unable to clean up messaging for process (%d)\n",
+			pid);
+		return NT_STATUS_NO_MEMORY;
+	}
+
+	names = irpc_all_servers(msg_ctx, mem_ctx);
+	if (names == NULL) {
+		TALLOC_FREE(mem_ctx);
+		return NT_STATUS_OK;
+	}
+	for (i = 0; i < names->num_records; i++) {
+		for (j = 0; j < names->names[i]->count; j++) {
+			if (names->names[i]->ids[j].pid == pid) {
+				int ret = server_id_db_prune_name(
+					msg_ctx->names,
+					names->names[i]->name,
+					names->names[i]->ids[j]);
+				if (ret != 0 && ret != ENOENT) {
+					TALLOC_FREE(mem_ctx);
+					return map_nt_error_from_unix_common(
+					    ret);
+				}
+			}
+		}
+	}
+	TALLOC_FREE(mem_ctx);
+	return NT_STATUS_OK;
+}
+
+static int imessaging_context_destructor(struct imessaging_context *msg)
+{
+	struct irpc_request *irpc = NULL;
+	struct irpc_request *next = NULL;
+
+	for (irpc = msg->requests; irpc != NULL; irpc = next) {
+		next = irpc->next;
+
+		DLIST_REMOVE(msg->requests, irpc);
+		irpc->callid = -1;
+	}
+
+	DLIST_REMOVE(msg_ctxs, msg);
+	TALLOC_FREE(msg->msg_dgm_ref);
+	return 0;
+}
+
+/*
+ * Cleanup messaging dgm contexts on a specific event context.
+ *
+ * We must make sure to unref all messaging_dgm_ref's *before* the
+ * tevent context goes away. Only when the last ref is freed, the
+ * refcounted messaging dgm context will be freed.
+ */
+void imessaging_dgm_unref_ev(struct tevent_context *ev)
+{
+	struct imessaging_context *msg = NULL;
+
+	for (msg = msg_ctxs; msg != NULL; msg = msg->next) {
+		if (msg->ev == ev) {
+			TALLOC_FREE(msg->msg_dgm_ref);
+		}
+	}
+}
+
+static NTSTATUS imessaging_reinit(struct imessaging_context *msg)
+{
+	int ret = -1;
+
+	TALLOC_FREE(msg->msg_dgm_ref);
+
+	if (msg->discard_incoming) {
+		msg->num_incoming_listeners = 0;
+	} else {
+		msg->num_incoming_listeners = 1;
+	}
+
+	msg->server_id.pid = getpid();
+
+	msg->msg_dgm_ref = messaging_dgm_ref(msg,
+				msg->ev,
+				&msg->server_id.unique_id,
+				msg->sock_dir,
+				msg->lock_dir,
+				imessaging_dgm_recv,
+				msg,
+				&ret);
+
+	if (msg->msg_dgm_ref == NULL) {
+		DEBUG(2, ("messaging_dgm_ref failed: %s\n",
+			strerror(ret)));
+		return map_nt_error_from_unix_common(ret);
+	}
+
+	server_id_db_reinit(msg->names, msg->server_id);
+	return NT_STATUS_OK;
+}
+
+/*
+ * Must be called after a fork.
+ */
+NTSTATUS imessaging_reinit_all(void)
+{
+	struct imessaging_context *msg = NULL;
+
+	for (msg = msg_ctxs; msg != NULL; msg = msg->next) {
+		NTSTATUS status = imessaging_reinit(msg);
+		if (!NT_STATUS_IS_OK(status)) {
+			return status;
+		}
+	}
+	return NT_STATUS_OK;
+}
+
+/*
+  create the listening socket and setup the dispatcher
+*/
+static struct imessaging_context *imessaging_init_internal(
+					   TALLOC_CTX *mem_ctx,
+					   bool discard_incoming,
+					   struct loadparm_context *lp_ctx,
+					   struct server_id server_id,
+					   struct tevent_context *ev)
+{
+	NTSTATUS status;
+	struct imessaging_context *msg;
+	bool ok;
+	int ret;
+	const char *lock_dir = NULL;
+	int tdb_flags = TDB_INCOMPATIBLE_HASH | TDB_CLEAR_IF_FIRST;
+
+	if (ev == NULL) {
+		return NULL;
+	}
+
+	msg = talloc_zero(mem_ctx, struct imessaging_context);
+	if (msg == NULL) {
+		return NULL;
+	}
+	msg->ev = ev;
+	msg->discard_incoming = discard_incoming;
+	if (msg->discard_incoming) {
+		msg->num_incoming_listeners = 0;
+	} else {
+		msg->num_incoming_listeners = 1;
+	}
+
+	talloc_set_destructor(msg, imessaging_context_destructor);
+
+	/* create the messaging directory if needed */
+
+	lock_dir = lpcfg_lock_directory(lp_ctx);
+	if (lock_dir == NULL) {
+		goto fail;
+	}
+
+	msg->sock_dir = lpcfg_private_path(msg, lp_ctx, "msg.sock");
+	if (msg->sock_dir == NULL) {
+		goto fail;
+	}
+	ok = directory_create_or_exist_strict(msg->sock_dir, geteuid(), 0700);
+	if (!ok) {
+		goto fail;
+	}
+
+	msg->lock_dir = lpcfg_lock_path(msg, lp_ctx, "msg.lock");
+	if (msg->lock_dir == NULL) {
+		goto fail;
+	}
+	ok = directory_create_or_exist_strict(msg->lock_dir, geteuid(), 0755);
+	if (!ok) {
+		goto fail;
+	}
+
+	msg->msg_dgm_ref = messaging_dgm_ref(
+		msg, ev, &server_id.unique_id, msg->sock_dir, msg->lock_dir,
+		imessaging_dgm_recv, msg, &ret);
+
+	if (msg->msg_dgm_ref == NULL) {
+		goto fail;
+	}
+
+	msg->server_id     = server_id;
+	msg->idr           = idr_init(msg);
+	if (msg->idr == NULL) {
+		goto fail;
+	}
+
+	msg->dispatch_tree = idr_init(msg);
+	if (msg->dispatch_tree == NULL) {
+		goto fail;
+	}
+
+	msg->start_time    = timeval_current();
+
+	tdb_flags |= lpcfg_tdb_flags(lp_ctx, 0);
+
+	/*
+	 * This context holds a destructor that cleans up any names
+	 * registered on this context on talloc_free()
+	 */
+	msg->names = server_id_db_init(msg, server_id, lock_dir, 0, tdb_flags);
+	if (msg->names == NULL) {
+		goto fail;
+	}
+
+	status = imessaging_register(msg, NULL, MSG_PING, ping_message);
+	if (!NT_STATUS_IS_OK(status)) {
+		goto fail;
+	}
+	status = imessaging_register(msg, NULL, MSG_REQ_POOL_USAGE,
+				     pool_message);
+	if (!NT_STATUS_IS_OK(status)) {
+		goto fail;
+	}
+	status = imessaging_register(msg, NULL, MSG_IRPC, irpc_handler);
+	if (!NT_STATUS_IS_OK(status)) {
+		goto fail;
+	}
+	status = imessaging_register(msg, NULL, MSG_REQ_RINGBUF_LOG,
+				     ringbuf_log_msg);
+	if (!NT_STATUS_IS_OK(status)) {
+		goto fail;
+	}
+	status = imessaging_register(msg, NULL, MSG_DEBUG,
+				     debug_imessage);
+	if (!NT_STATUS_IS_OK(status)) {
+		goto fail;
+	}
+	status = imessaging_register(msg, NULL, MSG_REQ_DEBUGLEVEL,
+				     debuglevel_imessage);
+	if (!NT_STATUS_IS_OK(status)) {
+		goto fail;
+	}
+	status = IRPC_REGISTER(msg, irpc, IRPC_UPTIME, irpc_uptime, msg);
+	if (!NT_STATUS_IS_OK(status)) {
+		goto fail;
+	}
+#if defined(DEVELOPER) || defined(ENABLE_SELFTEST)
+	/*
+	 * Register handlers for messages specific to developer and
+	 * self test builds
+	 */
+	status = imessaging_register_extra_handlers(msg);
+	if (!NT_STATUS_IS_OK(status)) {
+		goto fail;
+	}
+#endif /* defined(DEVELOPER) || defined(ENABLE_SELFTEST) */
+
+	DLIST_ADD(msg_ctxs, msg);
+
+	return msg;
+fail:
+	talloc_free(msg);
+	return NULL;
+}
+
+/*
+  create the listening socket and setup the dispatcher
+*/
+struct imessaging_context *imessaging_init(TALLOC_CTX *mem_ctx,
+					   struct loadparm_context *lp_ctx,
+					   struct server_id server_id,
+					   struct tevent_context *ev)
+{
+	bool discard_incoming = false;
+	return imessaging_init_internal(mem_ctx,
+					discard_incoming,
+					lp_ctx,
+					server_id,
+					ev);
+}
+
+struct imessaging_context *imessaging_init_discard_incoming(
+						TALLOC_CTX *mem_ctx,
+						struct loadparm_context *lp_ctx,
+						struct server_id server_id,
+						struct tevent_context *ev)
+{
+	bool discard_incoming = true;
+	return imessaging_init_internal(mem_ctx,
+					discard_incoming,
+					lp_ctx,
+					server_id,
+					ev);
+}
+
+struct imessaging_post_state {
+	struct imessaging_context *msg_ctx;
+	struct imessaging_post_state **busy_ref;
+	size_t buf_len;
+	uint8_t buf[];
+};
+
+static int imessaging_post_state_destructor(struct imessaging_post_state *state)
+{
+	if (state->busy_ref != NULL) {
+		*state->busy_ref = NULL;
+		state->busy_ref = NULL;
+	}
+	return 0;
+}
+
+static void imessaging_post_handler(struct tevent_context *ev,
+				    struct tevent_immediate *ti,
+				    void *private_data)
+{
+	struct imessaging_post_state *state = talloc_get_type_abort(
+		private_data, struct imessaging_post_state);
+
+	if (state == NULL) {
+		return;
+	}
+
+	/*
+	 * In usecases like using messaging_client_init() with irpc processing
+	 * we may free the imessaging_context during the messaging handler.
+	 * imessaging_post_state is a child of imessaging_context and
+	 * might be implicitly free'ed before the explicit TALLOC_FREE(state).
+	 *
+	 * The busy_ref pointer makes sure the destructor clears
+	 * the local 'state' variable.
+	 */
+
+	SMB_ASSERT(state->busy_ref == NULL);
+	state->busy_ref = &state;
+
+	imessaging_dgm_recv(ev, state->buf, state->buf_len, NULL, 0,
+			    state->msg_ctx);
+
+	state->busy_ref = NULL;
+	TALLOC_FREE(state);
+}
+
+static int imessaging_post_self(struct imessaging_context *msg,
+				const uint8_t *buf, size_t buf_len)
+{
+	struct tevent_immediate *ti;
+	struct imessaging_post_state *state;
+
+	state = talloc_size(
+		msg, offsetof(struct imessaging_post_state, buf) + buf_len);
+	if (state == NULL) {
+		return ENOMEM;
+	}
+	talloc_set_name_const(state, "struct imessaging_post_state");
+
+	talloc_set_destructor(state, imessaging_post_state_destructor);
+
+	ti = tevent_create_immediate(state);
+	if (ti == NULL) {
+		TALLOC_FREE(state);
+		return ENOMEM;
+	}
+
+	state->msg_ctx = msg;
+	state->busy_ref = NULL;
+	state->buf_len = buf_len;
+	memcpy(state->buf, buf, buf_len);
+
+	tevent_schedule_immediate(ti, msg->ev, imessaging_post_handler,
+				  state);
+
+	return 0;
+}
+
+static void imessaging_dgm_recv(struct tevent_context *ev,
+				const uint8_t *buf, size_t buf_len,
+				int *fds, size_t num_fds,
+				void *private_data)
+{
+	struct imessaging_context *msg = talloc_get_type_abort(
+		private_data, struct imessaging_context);
+	uint32_t msg_type;
+	struct server_id src, dst;
+	struct server_id_buf srcbuf, dstbuf;
+	DATA_BLOB data;
+
+	if (buf_len < MESSAGE_HDR_LENGTH) {
+		/* Invalid message, ignore */
+		return;
+	}
+
+	if (msg->num_incoming_listeners == 0) {
+		struct server_id_buf selfbuf;
+
+		message_hdr_get(&msg_type, &src, &dst, buf);
+
+		DBG_DEBUG("not listening - discarding message from "
+			  "src[%s] to dst[%s] (self[%s]) type=0x%x "
+			  "on %s event context\n",
+			   server_id_str_buf(src, &srcbuf),
+			   server_id_str_buf(dst, &dstbuf),
+			   server_id_str_buf(msg->server_id, &selfbuf),
+			   (unsigned)msg_type,
+			   (ev != msg->ev) ? "different" : "main");
+		return;
+	}
+
+	if (ev != msg->ev) {
+		int ret;
+		ret = imessaging_post_self(msg, buf, buf_len);
+		if (ret != 0) {
+			DBG_WARNING("imessaging_post_self failed: %s\n",
+				    strerror(ret));
+		}
+		return;
+	}
+
+	message_hdr_get(&msg_type, &src, &dst, buf);
+
+	data.data = discard_const_p(uint8_t, buf + MESSAGE_HDR_LENGTH);
+	data.length = buf_len - MESSAGE_HDR_LENGTH;
+
+	if ((cluster_id_equal(&dst, &msg->server_id)) ||
+	    ((dst.task_id == 0) && (msg->server_id.pid == 0))) {
+		struct dispatch_fn *d, *next;
+
+		DEBUG(10, ("%s: dst %s matches my id: %s, type=0x%x\n",
+			   __func__,
+			   server_id_str_buf(dst, &dstbuf),
+			   server_id_str_buf(msg->server_id, &srcbuf),
+			   (unsigned)msg_type));
+
+		d = imessaging_find_dispatch(msg, msg_type);
+
+		for (; d; d = next) {
+			next = d->next;
+			d->fn(msg,
+			      d->private_data,
+			      d->msg_type,
+			      src,
+			      num_fds,
+			      fds,
+			      &data);
+		}
+	} else {
+		DEBUG(10, ("%s: Ignoring type=0x%x dst %s, I am %s, \n",
+			   __func__, (unsigned)msg_type,
+			   server_id_str_buf(dst, &dstbuf),
+			   server_id_str_buf(msg->server_id, &srcbuf)));
+	}
+}
+
+/*
+   A hack, for the short term until we get 'client only' messaging in place
+*/
+struct imessaging_context *imessaging_client_init(TALLOC_CTX *mem_ctx,
+						  struct loadparm_context *lp_ctx,
+						struct tevent_context *ev)
+{
+	struct server_id id;
+	ZERO_STRUCT(id);
+	id.pid = getpid();
+	id.task_id = generate_random();
+	id.vnn = NONCLUSTER_VNN;
+
+	/* This is because we are not in the s3 serverid database */
+	id.unique_id = SERVERID_UNIQUE_ID_NOT_TO_VERIFY;
+
+	return imessaging_init_discard_incoming(mem_ctx, lp_ctx, id, ev);
+}
+
+/*
+  a list of registered irpc server functions
+*/
+struct irpc_list {
+	struct irpc_list *next, *prev;
+	struct GUID uuid;
+	const struct ndr_interface_table *table;
+	int callnum;
+	irpc_function_t fn;
+	void *private_data;
+};
+
+
+/*
+  register a irpc server function
+*/
+NTSTATUS irpc_register(struct imessaging_context *msg_ctx,
+		       const struct ndr_interface_table *table,
+		       int callnum, irpc_function_t fn, void *private_data)
+{
+	struct irpc_list *irpc;
+
+	/* override an existing handler, if any */
+	for (irpc=msg_ctx->irpc; irpc; irpc=irpc->next) {
+		if (irpc->table == table && irpc->callnum == callnum) {
+			break;
+		}
+	}
+	if (irpc == NULL) {
+		irpc = talloc(msg_ctx, struct irpc_list);
+		NT_STATUS_HAVE_NO_MEMORY(irpc);
+		DLIST_ADD(msg_ctx->irpc, irpc);
+	}
+
+	irpc->table   = table;
+	irpc->callnum = callnum;
+	irpc->fn      = fn;
+	irpc->private_data = private_data;
+	irpc->uuid = irpc->table->syntax_id.uuid;
+
+	return NT_STATUS_OK;
+}
+
+
+/*
+  handle an incoming irpc reply message
+*/
+static void irpc_handler_reply(struct imessaging_context *msg_ctx, struct irpc_message *m)
+{
+	struct irpc_request *irpc;
+
+	irpc = (struct irpc_request *)idr_find(msg_ctx->idr, m->header.callid);
+	if (irpc == NULL) return;
+
+	irpc->incoming.handler(irpc, m);
+}
+
+/*
+  send a irpc reply
+*/
+NTSTATUS irpc_send_reply(struct irpc_message *m, NTSTATUS status)
+{
+	struct ndr_push *push;
+	DATA_BLOB packet;
+	enum ndr_err_code ndr_err;
+
+	m->header.status = status;
+
+	/* setup the reply */
+	push = ndr_push_init_ctx(m->ndr);
+	if (push == NULL) {
+		status = NT_STATUS_NO_MEMORY;
+		goto failed;
+	}
+
+	m->header.flags |= IRPC_FLAG_REPLY;
+	m->header.creds.token= NULL;
+
+	/* construct the packet */
+	ndr_err = ndr_push_irpc_header(push, NDR_SCALARS|NDR_BUFFERS, &m->header);
+	if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
+		status = ndr_map_error2ntstatus(ndr_err);
+		goto failed;
+	}
+
+	ndr_err = m->irpc->table->calls[m->irpc->callnum].ndr_push(push, NDR_OUT, m->data);
+	if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
+		status = ndr_map_error2ntstatus(ndr_err);
+		goto failed;
+	}
+
+	/* send the reply message */
+	packet = ndr_push_blob(push);
+	status = imessaging_send(m->msg_ctx, m->from, MSG_IRPC, &packet);
+	if (!NT_STATUS_IS_OK(status)) goto failed;
+
+failed:
+	talloc_free(m);
+	return status;
+}
+
+/*
+  handle an incoming irpc request message
+*/
+static void irpc_handler_request(struct imessaging_context *msg_ctx,
+				 struct irpc_message *m)
+{
+	struct irpc_list *i;
+	void *r;
+	enum ndr_err_code ndr_err;
+
+	for (i=msg_ctx->irpc; i; i=i->next) {
+		if (GUID_equal(&i->uuid, &m->header.uuid) &&
+		    i->table->syntax_id.if_version == m->header.if_version &&
+		    i->callnum == m->header.callnum) {
+			break;
+		}
+	}
+
+	if (i == NULL) {
+		/* no registered handler for this message */
+		talloc_free(m);
+		return;
+	}
+
+	/* allocate space for the structure */
+	r = talloc_zero_size(m->ndr, i->table->calls[m->header.callnum].struct_size);
+	if (r == NULL) goto failed;
+
+	m->ndr->flags |= LIBNDR_FLAG_REF_ALLOC;
+
+	/* parse the request data */
+	ndr_err = i->table->calls[i->callnum].ndr_pull(m->ndr, NDR_IN, r);
+	if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) goto failed;
+
+	/* make the call */
+	m->private_data= i->private_data;
+	m->defer_reply = false;
+	m->no_reply    = false;
+	m->msg_ctx     = msg_ctx;
+	m->irpc        = i;
+	m->data        = r;
+
+	m->header.status = i->fn(m, r);
+
+	if (m->no_reply) {
+		/* the server function won't ever be replying to this request */
+		talloc_free(m);
+		return;
+	}
+
+	if (m->defer_reply) {
+		/* the server function has asked to defer the reply to later */
+		talloc_steal(msg_ctx, m);
+		return;
+	}
+
+	irpc_send_reply(m, m->header.status);
+	return;
+
+failed:
+	talloc_free(m);
+}
+
+/*
+  handle an incoming irpc message
+*/
+static void irpc_handler(struct imessaging_context *msg_ctx,
+			 void *private_data,
+			 uint32_t msg_type,
+			 struct server_id src,
+			 size_t num_fds,
+			 int *fds,
+			 DATA_BLOB *packet)
+{
+	struct irpc_message *m;
+	enum ndr_err_code ndr_err;
+
+	if (num_fds != 0) {
+		DBG_WARNING("Received %zu fds, ignoring message\n", num_fds);
+		return;
+	}
+
+	m = talloc(msg_ctx, struct irpc_message);
+	if (m == NULL) goto failed;
+
+	m->from = src;
+
+	m->ndr = ndr_pull_init_blob(packet, m);
+	if (m->ndr == NULL) goto failed;
+
+	m->ndr->flags |= LIBNDR_FLAG_REF_ALLOC;
+
+	ndr_err = ndr_pull_irpc_header(m->ndr, NDR_BUFFERS|NDR_SCALARS, &m->header);
+	if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) goto failed;
+
+	if (m->header.flags & IRPC_FLAG_REPLY) {
+		irpc_handler_reply(msg_ctx, m);
+	} else {
+		irpc_handler_request(msg_ctx, m);
+	}
+	return;
+
+failed:
+	talloc_free(m);
+}
+
+
+/*
+  destroy a irpc request
+*/
+static int irpc_destructor(struct irpc_request *irpc)
+{
+	if (irpc->callid != -1) {
+		DLIST_REMOVE(irpc->msg_ctx->requests, irpc);
+		idr_remove(irpc->msg_ctx->idr, irpc->callid);
+		if (irpc->msg_ctx->discard_incoming) {
+			SMB_ASSERT(irpc->msg_ctx->num_incoming_listeners > 0);
+		} else {
+			SMB_ASSERT(irpc->msg_ctx->num_incoming_listeners > 1);
+		}
+		irpc->msg_ctx->num_incoming_listeners -= 1;
+		irpc->callid = -1;
+	}
+
+	return 0;
+}
+
+/*
+  add a string name that this irpc server can be called on
+
+  It will be removed from the DB either via irpc_remove_name or on
+  talloc_free(msg_ctx->names).
+*/
+NTSTATUS irpc_add_name(struct imessaging_context *msg_ctx, const char *name)
+{
+	int ret;
+
+	ret = server_id_db_add(msg_ctx->names, name);
+	if (ret != 0) {
+		return map_nt_error_from_unix_common(ret);
+	}
+	return NT_STATUS_OK;
+}
+
+static int all_servers_func(const char *name, unsigned num_servers,
+			    const struct server_id *servers,
+			    void *private_data)
+{
+	struct irpc_name_records *name_records = talloc_get_type(
+		private_data, struct irpc_name_records);
+	struct irpc_name_record *name_record;
+	uint32_t i;
+
+	name_records->names
+		= talloc_realloc(name_records, name_records->names,
+				 struct irpc_name_record *, name_records->num_records+1);
+	if (!name_records->names) {
+		return -1;
+	}
+
+	name_records->names[name_records->num_records] = name_record
+		= talloc(name_records->names,
+			 struct irpc_name_record);
+	if (!name_record) {
+		return -1;
+	}
+
+	name_records->num_records++;
+
+	name_record->name = talloc_strdup(name_record, name);
+	if (!name_record->name) {
+		return -1;
+	}
+
+	name_record->count = num_servers;
+	name_record->ids = talloc_array(name_record, struct server_id,
+					num_servers);
+	if (name_record->ids == NULL) {
+		return -1;
+	}
+	for (i=0;i<name_record->count;i++) {
+		name_record->ids[i] = servers[i];
+	}
+	return 0;
+}
+
+/*
+  return a list of server ids for a server name
+*/
+struct irpc_name_records *irpc_all_servers(struct imessaging_context *msg_ctx,
+					   TALLOC_CTX *mem_ctx)
+{
+	int ret;
+	struct irpc_name_records *name_records = talloc_zero(mem_ctx, struct irpc_name_records);
+	if (name_records == NULL) {
+		return NULL;
+	}
+
+	ret = server_id_db_traverse_read(msg_ctx->names, all_servers_func,
+					 name_records);
+	if (ret == -1) {
+		TALLOC_FREE(name_records);
+		return NULL;
+	}
+
+	return name_records;
+}
+
+/*
+  remove a name from a messaging context
+*/
+void irpc_remove_name(struct imessaging_context *msg_ctx, const char *name)
+{
+	server_id_db_remove(msg_ctx->names, name);
+}
+
+struct server_id imessaging_get_server_id(struct imessaging_context *msg_ctx)
+{
+	return msg_ctx->server_id;
+}
+
+struct irpc_bh_state {
+	struct imessaging_context *msg_ctx;
+	struct server_id server_id;
+	const struct ndr_interface_table *table;
+	uint32_t timeout;
+	struct security_token *token;
+};
+
+static bool irpc_bh_is_connected(struct dcerpc_binding_handle *h)
+{
+	struct irpc_bh_state *hs = dcerpc_binding_handle_data(h,
+				   struct irpc_bh_state);
+
+	if (!hs->msg_ctx) {
+		return false;
+	}
+
+	return true;
+}
+
+static uint32_t irpc_bh_set_timeout(struct dcerpc_binding_handle *h,
+				    uint32_t timeout)
+{
+	struct irpc_bh_state *hs = dcerpc_binding_handle_data(h,
+				   struct irpc_bh_state);
+	uint32_t old = hs->timeout;
+
+	hs->timeout = timeout;
+
+	return old;
+}
+
+struct irpc_bh_raw_call_state {
+	struct irpc_request *irpc;
+	uint32_t opnum;
+	DATA_BLOB in_data;
+	DATA_BLOB in_packet;
+	DATA_BLOB out_data;
+};
+
+static void irpc_bh_raw_call_incoming_handler(struct irpc_request *irpc,
+					      struct irpc_message *m);
+
+static struct tevent_req *irpc_bh_raw_call_send(TALLOC_CTX *mem_ctx,
+						struct tevent_context *ev,
+						struct dcerpc_binding_handle *h,
+						const struct GUID *object,
+						uint32_t opnum,
+						uint32_t in_flags,
+						const uint8_t *in_data,
+						size_t in_length)
+{
+	struct irpc_bh_state *hs =
+		dcerpc_binding_handle_data(h,
+		struct irpc_bh_state);
+	struct tevent_req *req;
+	struct irpc_bh_raw_call_state *state;
+	bool ok;
+	struct irpc_header header;
+	struct ndr_push *ndr;
+	NTSTATUS status;
+	enum ndr_err_code ndr_err;
+
+	req = tevent_req_create(mem_ctx, &state,
+				struct irpc_bh_raw_call_state);
+	if (req == NULL) {
+		return NULL;
+	}
+	state->opnum = opnum;
+	state->in_data.data = discard_const_p(uint8_t, in_data);
+	state->in_data.length = in_length;
+
+	ok = irpc_bh_is_connected(h);
+	if (!ok) {
+		tevent_req_nterror(req, NT_STATUS_CONNECTION_DISCONNECTED);
+		return tevent_req_post(req, ev);
+	}
+
+	state->irpc = talloc_zero(state, struct irpc_request);
+	if (tevent_req_nomem(state->irpc, req)) {
+		return tevent_req_post(req, ev);
+	}
+
+	state->irpc->msg_ctx  = hs->msg_ctx;
+	state->irpc->callid   = idr_get_new(hs->msg_ctx->idr,
+					    state->irpc, UINT16_MAX);
+	if (state->irpc->callid == -1) {
+		tevent_req_nterror(req, NT_STATUS_INSUFFICIENT_RESOURCES);
+		return tevent_req_post(req, ev);
+	}
+	state->irpc->incoming.handler = irpc_bh_raw_call_incoming_handler;
+	state->irpc->incoming.private_data = req;
+
+	/* make sure we accept incoming messages */
+	SMB_ASSERT(state->irpc->msg_ctx->num_incoming_listeners < UINT64_MAX);
+	state->irpc->msg_ctx->num_incoming_listeners += 1;
+	DLIST_ADD_END(state->irpc->msg_ctx->requests, state->irpc);
+	talloc_set_destructor(state->irpc, irpc_destructor);
+
+	/* setup the header */
+	header.uuid = hs->table->syntax_id.uuid;
+
+	header.if_version = hs->table->syntax_id.if_version;
+	header.callid     = state->irpc->callid;
+	header.callnum    = state->opnum;
+	header.flags      = 0;
+	header.status     = NT_STATUS_OK;
+	header.creds.token= hs->token;
+
+	/* construct the irpc packet */
+	ndr = ndr_push_init_ctx(state->irpc);
+	if (tevent_req_nomem(ndr, req)) {
+		return tevent_req_post(req, ev);
+	}
+
+	ndr_err = ndr_push_irpc_header(ndr, NDR_SCALARS|NDR_BUFFERS, &header);
+	status = ndr_map_error2ntstatus(ndr_err);
+	if (!NT_STATUS_IS_OK(status)) {
+		tevent_req_nterror(req, status);
+		return tevent_req_post(req, ev);
+	}
+
+	ndr_err = ndr_push_bytes(ndr, in_data, in_length);
+	status = ndr_map_error2ntstatus(ndr_err);
+	if (!NT_STATUS_IS_OK(status)) {
+		tevent_req_nterror(req, status);
+		return tevent_req_post(req, ev);
+	}
+
+	/* and send it */
+	state->in_packet = ndr_push_blob(ndr);
+	status = imessaging_send(hs->msg_ctx, hs->server_id,
+				MSG_IRPC, &state->in_packet);
+	if (!NT_STATUS_IS_OK(status)) {
+		tevent_req_nterror(req, status);
+		return tevent_req_post(req, ev);
+	}
+
+	if (hs->timeout != IRPC_CALL_TIMEOUT_INF) {
+		/* set timeout-callback in case caller wants that */
+		ok = tevent_req_set_endtime(req, ev, timeval_current_ofs(hs->timeout, 0));
+		if (!ok) {
+			return tevent_req_post(req, ev);
+		}
+	}
+
+	return req;
+}
+
+static void irpc_bh_raw_call_incoming_handler(struct irpc_request *irpc,
+					      struct irpc_message *m)
+{
+	struct tevent_req *req =
+		talloc_get_type_abort(irpc->incoming.private_data,
+		struct tevent_req);
+	struct irpc_bh_raw_call_state *state =
+		tevent_req_data(req,
+		struct irpc_bh_raw_call_state);
+
+	talloc_steal(state, m);
+
+	if (!NT_STATUS_IS_OK(m->header.status)) {
+		tevent_req_nterror(req, m->header.status);
+		return;
+	}
+
+	state->out_data = data_blob_talloc(state,
+		m->ndr->data + m->ndr->offset,
+		m->ndr->data_size - m->ndr->offset);
+	if ((m->ndr->data_size - m->ndr->offset) > 0 && !state->out_data.data) {
+		tevent_req_oom(req);
+		return;
+	}
+
+	tevent_req_done(req);
+}
+
+static NTSTATUS irpc_bh_raw_call_recv(struct tevent_req *req,
+					TALLOC_CTX *mem_ctx,
+					uint8_t **out_data,
+					size_t *out_length,
+					uint32_t *out_flags)
+{
+	struct irpc_bh_raw_call_state *state =
+		tevent_req_data(req,
+		struct irpc_bh_raw_call_state);
+	NTSTATUS status;
+
+	if (tevent_req_is_nterror(req, &status)) {
+		tevent_req_received(req);
+		return status;
+	}
+
+	*out_data = talloc_move(mem_ctx, &state->out_data.data);
+	*out_length = state->out_data.length;
+	*out_flags = 0;
+	tevent_req_received(req);
+	return NT_STATUS_OK;
+}
+
+struct irpc_bh_disconnect_state {
+	uint8_t _dummy;
+};
+
+static struct tevent_req *irpc_bh_disconnect_send(TALLOC_CTX *mem_ctx,
+						struct tevent_context *ev,
+						struct dcerpc_binding_handle *h)
+{
+	struct irpc_bh_state *hs = dcerpc_binding_handle_data(h,
+				     struct irpc_bh_state);
+	struct tevent_req *req;
+	struct irpc_bh_disconnect_state *state;
+	bool ok;
+
+	req = tevent_req_create(mem_ctx, &state,
+				struct irpc_bh_disconnect_state);
+	if (req == NULL) {
+		return NULL;
+	}
+
+	ok = irpc_bh_is_connected(h);
+	if (!ok) {
+		tevent_req_nterror(req, NT_STATUS_CONNECTION_DISCONNECTED);
+		return tevent_req_post(req, ev);
+	}
+
+	hs->msg_ctx = NULL;
+
+	tevent_req_done(req);
+	return tevent_req_post(req, ev);
+}
+
+static NTSTATUS irpc_bh_disconnect_recv(struct tevent_req *req)
+{
+	NTSTATUS status;
+
+	if (tevent_req_is_nterror(req, &status)) {
+		tevent_req_received(req);
+		return status;
+	}
+
+	tevent_req_received(req);
+	return NT_STATUS_OK;
+}
+
+static bool irpc_bh_ref_alloc(struct dcerpc_binding_handle *h)
+{
+	return true;
+}
+
+static const struct dcerpc_binding_handle_ops irpc_bh_ops = {
+	.name			= "wbint",
+	.is_connected		= irpc_bh_is_connected,
+	.set_timeout		= irpc_bh_set_timeout,
+	.raw_call_send		= irpc_bh_raw_call_send,
+	.raw_call_recv		= irpc_bh_raw_call_recv,
+	.disconnect_send	= irpc_bh_disconnect_send,
+	.disconnect_recv	= irpc_bh_disconnect_recv,
+
+	.ref_alloc		= irpc_bh_ref_alloc,
+};
+
+/* initialise a irpc binding handle */
+struct dcerpc_binding_handle *irpc_binding_handle(TALLOC_CTX *mem_ctx,
+						  struct imessaging_context *msg_ctx,
+						  struct server_id server_id,
+						  const struct ndr_interface_table *table)
+{
+	struct dcerpc_binding_handle *h;
+	struct irpc_bh_state *hs;
+
+	h = dcerpc_binding_handle_create(mem_ctx,
+					 &irpc_bh_ops,
+					 NULL,
+					 table,
+					 &hs,
+					 struct irpc_bh_state,
+					 __location__);
+	if (h == NULL) {
+		return NULL;
+	}
+	hs->msg_ctx = msg_ctx;
+	hs->server_id = server_id;
+	hs->table = table;
+	hs->timeout = IRPC_CALL_TIMEOUT;
+
+	return h;
+}
+
+struct dcerpc_binding_handle *irpc_binding_handle_by_name(TALLOC_CTX *mem_ctx,
+							  struct imessaging_context *msg_ctx,
+							  const char *dest_task,
+							  const struct ndr_interface_table *table)
+{
+	struct dcerpc_binding_handle *h;
+	unsigned num_sids;
+	struct server_id *sids;
+	struct server_id sid;
+	NTSTATUS status;
+
+	/* find the server task */
+
+	status = irpc_servers_byname(msg_ctx, mem_ctx, dest_task,
+				     &num_sids, &sids);
+	if (!NT_STATUS_IS_OK(status)) {
+		errno = EADDRNOTAVAIL;
+		return NULL;
+	}
+	sid = sids[0];
+	talloc_free(sids);
+
+	h = irpc_binding_handle(mem_ctx, msg_ctx,
+				sid, table);
+	if (h == NULL) {
+		return NULL;
+	}
+
+	return h;
+}
+
+void irpc_binding_handle_add_security_token(struct dcerpc_binding_handle *h,
+					    struct security_token *token)
+{
+	struct irpc_bh_state *hs =
+		dcerpc_binding_handle_data(h,
+		struct irpc_bh_state);
+
+	hs->token = token;
+}
diff --git a/source4/lib/messaging/messaging.h b/source4/lib/messaging/messaging.h
new file mode 100644
index 0000000..e7ae9e8
--- /dev/null
+++ b/source4/lib/messaging/messaging.h
@@ -0,0 +1,70 @@
+/* 
+   Unix SMB/CIFS implementation.
+   messages.c header
+   Copyright (C) Andrew Tridgell 2000
+   Copyright (C) 2001, 2002 by Martin Pool
+   
+   This program is free software; you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation; either version 3 of the License, or
+   (at your option) any later version.
+   
+   This program 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 General Public License for more details.
+   
+   You should have received a copy of the GNU General Public License
+   along with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#ifndef _SOURCE4_LIB_MESSAGING_MESSAGES_H_
+#define _SOURCE4_LIB_MESSAGING_MESSAGES_H_
+
+#include "librpc/gen_ndr/server_id.h"
+#include "librpc/gen_ndr/messaging.h"
+
+struct imessaging_context;
+
+/* taskid for messaging of parent process */
+#define SAMBA_PARENT_TASKID     0
+
+typedef void (*msg_callback_t)(
+	struct imessaging_context *msg,
+	void *private_data,
+	uint32_t msg_type,
+	struct server_id server_id,
+	size_t num_fds,
+	int *fds,
+	DATA_BLOB *data);
+
+NTSTATUS imessaging_send(struct imessaging_context *msg, struct server_id server,
+			uint32_t msg_type, const DATA_BLOB *data);
+NTSTATUS imessaging_register(struct imessaging_context *msg, void *private_data,
+			    uint32_t msg_type,
+			    msg_callback_t fn);
+NTSTATUS imessaging_register_tmp(struct imessaging_context *msg, void *private_data,
+				msg_callback_t fn, uint32_t *msg_type);
+struct imessaging_context *imessaging_init(TALLOC_CTX *mem_ctx,
+					   struct loadparm_context *lp_ctx,
+					   struct server_id server_id,
+					   struct tevent_context *ev);
+struct imessaging_context *imessaging_init_discard_incoming(
+						TALLOC_CTX *mem_ctx,
+						struct loadparm_context *lp_ctx,
+						struct server_id server_id,
+						struct tevent_context *ev);
+void imessaging_dgm_unref_ev(struct tevent_context *ev);
+NTSTATUS imessaging_reinit_all(void);
+int imessaging_cleanup(struct imessaging_context *msg);
+struct imessaging_context *imessaging_client_init(TALLOC_CTX *mem_ctx,
+					   struct loadparm_context *lp_ctx,
+					 struct tevent_context *ev);
+NTSTATUS imessaging_send_ptr(struct imessaging_context *msg, struct server_id server,
+			    uint32_t msg_type, void *ptr);
+void imessaging_deregister(struct imessaging_context *msg, uint32_t msg_type, void *private_data);
+struct server_id imessaging_get_server_id(struct imessaging_context *msg_ctx);
+NTSTATUS imessaging_process_cleanup(struct imessaging_context *msg_ctx,
+				    pid_t pid);
+
+#endif
diff --git a/source4/lib/messaging/messaging_handlers.c b/source4/lib/messaging/messaging_handlers.c
new file mode 100644
index 0000000..57e3e1c
--- /dev/null
+++ b/source4/lib/messaging/messaging_handlers.c
@@ -0,0 +1,135 @@
+/*
+   Unix SMB/CIFS implementation.
+
+   Handers for non core Samba internal messages
+
+   Handlers for messages that are only included in developer and self test
+   builds.
+
+   Copyright (C) Andrew Bartlett <abartlet@samba.org> 2018
+
+   This program is free software; you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation; either version 3 of the License, or
+   (at your option) any later version.
+
+   This program 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 General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#include "includes.h"
+#include "lib/util/server_id.h"
+#include "messaging/messaging.h"
+#include "messaging/messaging_internal.h"
+
+#if defined(DEVELOPER) || defined(ENABLE_SELFTEST)
+
+/*
+ * Inject a fault into the currently running process
+ */
+static void do_inject_fault(struct imessaging_context *msg,
+			    void *private_data,
+			    uint32_t msg_type,
+			    struct server_id src,
+			    size_t num_fds,
+			    int *fds,
+			    DATA_BLOB *data)
+{
+	int sig;
+	struct server_id_buf tmp;
+
+	if (num_fds != 0) {
+		DBG_WARNING("Received %zu fds, ignoring message\n", num_fds);
+		return;
+	}
+
+	if (data->length != sizeof(sig)) {
+		DBG_ERR("Process %s sent bogus signal injection request\n",
+			server_id_str_buf(src, &tmp));
+		return;
+	}
+
+	sig = *(int *)data->data;
+	if (sig == -1) {
+		DBG_ERR("Process %s requested an iternal failure, "
+			"calling exit(1)\n",
+			server_id_str_buf(src, &tmp));
+		exit(1);
+	}
+
+#if HAVE_STRSIGNAL
+	DBG_ERR("Process %s requested injection of signal %d (%s)\n",
+		server_id_str_buf(src, &tmp),
+		sig,
+		strsignal(sig));
+#else
+	DBG_ERR("Process %s requested injection of signal %d\n",
+		server_id_str_buf(src, &tmp),
+		sig);
+#endif
+
+	kill(getpid(), sig);
+}
+
+/*
+ * Cause the current process to sleep for a specified number of seconds
+ */
+static void do_sleep(struct imessaging_context *msg,
+		     void *private_data,
+		     uint32_t msg_type,
+		     struct server_id src,
+		     size_t num_fds,
+		     int *fds,
+		     DATA_BLOB *data)
+{
+	unsigned int seconds;
+	struct server_id_buf tmp;
+
+	if (num_fds != 0) {
+		DBG_WARNING("Received %zu fds, ignoring message\n", num_fds);
+		return;
+	}
+
+	if (data->length != sizeof(seconds)) {
+		DBG_ERR("Process %s sent bogus sleep request\n",
+			server_id_str_buf(src, &tmp));
+		return;
+	}
+
+	seconds = *(unsigned int *)data->data;
+	DBG_ERR("Process %s requested a sleep of %u seconds\n",
+		server_id_str_buf(src, &tmp),
+		seconds);
+	sleep(seconds);
+	DBG_ERR("Restarting after %u second sleep requested by process %s\n",
+		seconds,
+		server_id_str_buf(src, &tmp));
+}
+
+/*
+ * Register the extra messaging handlers
+ */
+NTSTATUS imessaging_register_extra_handlers(struct imessaging_context *msg)
+{
+	NTSTATUS status;
+
+	status = imessaging_register(
+	    msg, NULL, MSG_SMB_INJECT_FAULT, do_inject_fault);
+	if (!NT_STATUS_IS_OK(status)) {
+		return status;
+	}
+
+	status = imessaging_register(msg, NULL, MSG_SMB_SLEEP, do_sleep);
+	if (!NT_STATUS_IS_OK(status)) {
+		return status;
+	}
+
+	return NT_STATUS_OK;
+}
+
+#endif /* defined(DEVELOPER) || defined(ENABLE_SELFTEST) */
diff --git a/source4/lib/messaging/messaging_internal.h b/source4/lib/messaging/messaging_internal.h
new file mode 100644
index 0000000..6281bda
--- /dev/null
+++ b/source4/lib/messaging/messaging_internal.h
@@ -0,0 +1,50 @@
+/*
+   Unix SMB/CIFS implementation.
+
+   Samba internal messaging functions
+
+   Copyright (C) Andrew Tridgell 2004
+
+   This program is free software; you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation; either version 3 of the License, or
+   (at your option) any later version.
+
+   This program 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 General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+struct irpc_request;
+
+struct imessaging_context {
+	struct imessaging_context *prev, *next;
+	struct tevent_context *ev;
+	struct server_id server_id;
+	const char *sock_dir;
+	const char *lock_dir;
+	struct dispatch_fn **dispatch;
+	uint32_t num_types;
+	struct idr_context *dispatch_tree;
+	struct irpc_list *irpc;
+	struct idr_context *idr;
+	struct irpc_request *requests;
+	struct server_id_db *names;
+	struct timeval start_time;
+	void *msg_dgm_ref;
+	/*
+	 * The number of instances waiting for incoming
+	 * messages. By default it's always greater than 0.
+	 *
+	 * If it's 0 we'll discard incoming messages,
+	 * see imessaging_init_discard_imcoming().
+	 */
+	bool discard_incoming;
+	uint64_t num_incoming_listeners;
+};
+
+NTSTATUS imessaging_register_extra_handlers(struct imessaging_context *msg);
diff --git a/source4/lib/messaging/messaging_send.c b/source4/lib/messaging/messaging_send.c
new file mode 100644
index 0000000..24cdce3
--- /dev/null
+++ b/source4/lib/messaging/messaging_send.c
@@ -0,0 +1,115 @@
+/*
+   Unix SMB/CIFS implementation.
+
+   Samba internal messaging functions (send).
+
+   Copyright (C) Andrew Tridgell 2004
+
+   This program is free software; you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation; either version 3 of the License, or
+   (at your option) any later version.
+
+   This program 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 General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#include "includes.h"
+#include "messaging/messaging.h"
+#include "messaging/irpc.h"
+#include "lib/messaging/messages_dgm.h"
+#include "lib/messaging/messages_dgm_ref.h"
+#include "../source3/lib/messages_util.h"
+#include "messaging/messaging_internal.h"
+#include "lib/util/server_id_db.h"
+#include "cluster/cluster.h"
+#include "../lib/util/unix_privs.h"
+
+/*
+ * This file is for functions that can be called from auth_log without
+ * depending on all of dcerpc and so cause dep loops.
+ */
+
+/*
+  return a list of server ids for a server name
+*/
+NTSTATUS irpc_servers_byname(struct imessaging_context *msg_ctx,
+			     TALLOC_CTX *mem_ctx, const char *name,
+			     unsigned *num_servers,
+			     struct server_id **servers)
+{
+	int ret;
+
+	ret = server_id_db_lookup(msg_ctx->names, name, mem_ctx,
+				  num_servers, servers);
+	if (ret != 0) {
+		return map_nt_error_from_unix_common(ret);
+	}
+	return NT_STATUS_OK;
+}
+
+/*
+  Send a message to a particular server
+*/
+NTSTATUS imessaging_send(struct imessaging_context *msg, struct server_id server,
+			uint32_t msg_type, const DATA_BLOB *data)
+{
+	uint8_t hdr[MESSAGE_HDR_LENGTH];
+	struct iovec iov[2];
+	int num_iov, ret;
+	pid_t pid;
+	void *priv;
+
+	if (!cluster_node_equal(&msg->server_id, &server)) {
+		/* No cluster in source4... */
+		return NT_STATUS_OK;
+	}
+
+	message_hdr_put(hdr, msg_type, msg->server_id, server);
+
+	iov[0] = (struct iovec) { .iov_base = &hdr, .iov_len = sizeof(hdr) };
+	num_iov = 1;
+
+	if (data != NULL) {
+		iov[1] = (struct iovec) { .iov_base = data->data,
+					  .iov_len = data->length };
+		num_iov += 1;
+	}
+
+	pid = server.pid;
+	if (pid == 0) {
+		pid = getpid();
+	}
+
+	ret = messaging_dgm_send(pid, iov, num_iov, NULL, 0);
+
+	if (ret == EACCES) {
+		priv = root_privileges();
+		ret = messaging_dgm_send(pid, iov, num_iov, NULL, 0);
+		TALLOC_FREE(priv);
+	}
+
+	if (ret != 0) {
+		return map_nt_error_from_unix_common(ret);
+	}
+	return NT_STATUS_OK;
+}
+
+/*
+  Send a message to a particular server, with the message containing a single pointer
+*/
+NTSTATUS imessaging_send_ptr(struct imessaging_context *msg, struct server_id server,
+			    uint32_t msg_type, void *ptr)
+{
+	DATA_BLOB blob;
+
+	blob.data = (uint8_t *)&ptr;
+	blob.length = sizeof(void *);
+
+	return imessaging_send(msg, server, msg_type, &blob);
+}
diff --git a/source4/lib/messaging/pymessaging.c b/source4/lib/messaging/pymessaging.c
new file mode 100644
index 0000000..148c941
--- /dev/null
+++ b/source4/lib/messaging/pymessaging.c
@@ -0,0 +1,551 @@
+/* 
+   Unix SMB/CIFS implementation.
+   Copyright © Jelmer Vernooij <jelmer@samba.org> 2008
+
+   Based on the equivalent for EJS:
+   Copyright © Andrew Tridgell <tridge@samba.org> 2005
+   
+   This program is free software; you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation; either version 3 of the License, or
+   (at your option) any later version.
+   
+   This program 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 General Public License for more details.
+   
+   You should have received a copy of the GNU General Public License
+   along with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#include <Python.h>
+#include "python/py3compat.h"
+#include "includes.h"
+#include "python/modules.h"
+#include "libcli/util/pyerrors.h"
+#include "librpc/rpc/pyrpc_util.h"
+#include "librpc/ndr/libndr.h"
+#include "lib/messaging/messaging.h"
+#include "lib/messaging/irpc.h"
+#include "lib/events/events.h"
+#include "cluster/cluster.h"
+#include "param/param.h"
+#include "param/pyparam.h"
+#include "librpc/rpc/dcerpc.h"
+#include "librpc/gen_ndr/server_id.h"
+#include <pytalloc.h>
+#include "messaging_internal.h"
+
+
+extern PyTypeObject imessaging_Type;
+
+static bool server_id_from_py(PyObject *object, struct server_id *server_id)
+{
+	if (!PyTuple_Check(object)) {
+		if (!py_check_dcerpc_type(object, "samba.dcerpc.server_id", "server_id")) {
+
+			PyErr_SetString(PyExc_ValueError, "Expected tuple or server_id");
+			return false;
+		}
+		*server_id = *pytalloc_get_type(object, struct server_id);
+		return true;
+	}
+	if (PyTuple_Size(object) == 3) {
+		unsigned long long pid;
+		int task_id, vnn;
+
+		if (!PyArg_ParseTuple(object, "Kii", &pid, &task_id, &vnn)) {
+			return false;
+		}
+		server_id->pid = pid;
+		server_id->task_id = task_id;
+		server_id->vnn = vnn;
+		return true;
+	} else if (PyTuple_Size(object) == 2) {
+		unsigned long long pid;
+		int task_id;
+		if (!PyArg_ParseTuple(object, "Ki", &pid, &task_id))
+			return false;
+		*server_id = cluster_id(pid, task_id);
+		return true;
+	} else {
+		unsigned long long pid = getpid();
+		int task_id;
+		if (!PyArg_ParseTuple(object, "i", &task_id))
+			return false;
+		*server_id = cluster_id(pid, task_id);
+		return true;
+	}
+}
+
+typedef struct {
+	PyObject_HEAD
+	TALLOC_CTX *mem_ctx;
+	struct imessaging_context *msg_ctx;
+} imessaging_Object;
+
+static PyObject *py_imessaging_connect(PyTypeObject *self, PyObject *args, PyObject *kwargs)
+{
+	struct tevent_context *ev;
+	const char *kwnames[] = { "own_id", "lp_ctx", NULL };
+	PyObject *own_id = Py_None;
+	PyObject *py_lp_ctx = Py_None;
+	imessaging_Object *ret;
+	struct loadparm_context *lp_ctx;
+
+	if (!PyArg_ParseTupleAndKeywords(args, kwargs, "|OO:connect", 
+		discard_const_p(char *, kwnames), &own_id, &py_lp_ctx)) {
+		return NULL;
+	}
+
+	ret = PyObject_New(imessaging_Object, &imessaging_Type);
+	if (ret == NULL)
+		return NULL;
+
+	ret->mem_ctx = talloc_new(NULL);
+
+	lp_ctx = lpcfg_from_py_object(ret->mem_ctx, py_lp_ctx);
+	if (lp_ctx == NULL) {
+		PyErr_SetString(PyExc_RuntimeError, "imessaging_connect unable to interpret loadparm_context");
+		talloc_free(ret->mem_ctx);
+		return NULL;
+	}
+
+	ev = s4_event_context_init(ret->mem_ctx);
+
+	if (own_id != Py_None) {
+		struct server_id server_id;
+
+		if (!server_id_from_py(own_id, &server_id)) 
+			return NULL;
+
+		ret->msg_ctx = imessaging_init(ret->mem_ctx,
+					       lp_ctx,
+					       server_id,
+					       ev);
+	} else {
+		ret->msg_ctx = imessaging_client_init(ret->mem_ctx,
+						      lp_ctx,
+						      ev);
+	}
+
+	if (ret->msg_ctx == NULL) {
+		PyErr_SetString(PyExc_RuntimeError, "imessaging_connect unable to create a messaging context");
+		talloc_free(ret->mem_ctx);
+		return NULL;
+	}
+
+	return (PyObject *)ret;
+}
+
+static void py_imessaging_dealloc(PyObject *self)
+{
+	imessaging_Object *iface = (imessaging_Object *)self;
+	talloc_free(iface->msg_ctx);
+	self->ob_type->tp_free(self);
+}
+
+static PyObject *py_imessaging_send(PyObject *self, PyObject *args, PyObject *kwargs)
+{
+	imessaging_Object *iface = (imessaging_Object *)self;
+	uint32_t msg_type;
+	DATA_BLOB data;
+	PyObject *target;
+	NTSTATUS status;
+	struct server_id server;
+	const char *kwnames[] = { "target", "msg_type", "data", NULL };
+	Py_ssize_t length;
+
+	if (!PyArg_ParseTupleAndKeywords(args, kwargs, "OIs#:send",
+		discard_const_p(char *, kwnames), &target, &msg_type, &data.data, &length)) {
+
+		return NULL;
+	}
+
+	data.length = length;
+
+	if (!server_id_from_py(target, &server))
+		return NULL;
+
+	status = imessaging_send(iface->msg_ctx, server, msg_type, &data);
+	if (NT_STATUS_IS_ERR(status)) {
+		PyErr_SetNTSTATUS(status);
+		return NULL;
+	}
+
+	Py_RETURN_NONE;
+}
+
+static void py_msg_callback_wrapper(struct imessaging_context *msg,
+				    void *private_data,
+				    uint32_t msg_type,
+				    struct server_id server_id,
+				    size_t num_fds,
+				    int *fds,
+				    DATA_BLOB *data)
+{
+	PyObject *py_server_id, *callback_and_tuple = (PyObject *)private_data;
+	PyObject *callback, *py_private;
+
+	struct server_id *p_server_id = talloc(NULL, struct server_id);
+
+	if (num_fds != 0) {
+		DBG_WARNING("Received %zu fds, ignoring message\n", num_fds);
+		return;
+	}
+
+	if (!p_server_id) {
+		PyErr_NoMemory();
+		return;
+	}
+	*p_server_id = server_id;
+
+	if (!PyArg_ParseTuple(callback_and_tuple, "OO",
+			      &callback,
+			      &py_private)) {
+		return;
+	}
+
+	py_server_id = py_return_ndr_struct("samba.dcerpc.server_id", "server_id", p_server_id, p_server_id);
+	talloc_unlink(NULL, p_server_id);
+
+	PyObject_CallFunction(callback, discard_const_p(char, "OiOs#"),
+			      py_private,
+			      msg_type,
+			      py_server_id,
+			      data->data, data->length);
+}
+
+static PyObject *py_imessaging_register(PyObject *self, PyObject *args, PyObject *kwargs)
+{
+	imessaging_Object *iface = (imessaging_Object *)self;
+	int msg_type = -1;
+	PyObject *callback_and_context;
+	NTSTATUS status;
+	const char *kwnames[] = { "callback_and_context", "msg_type", NULL };
+	
+	if (!PyArg_ParseTupleAndKeywords(args, kwargs, "O|i:register", 
+		discard_const_p(char *, kwnames),
+					 &callback_and_context, &msg_type)) {
+		return NULL;
+	}
+	if (!PyTuple_Check(callback_and_context)
+	    || PyTuple_Size(callback_and_context) != 2) {
+		PyErr_SetString(PyExc_ValueError, "Expected of size 2 for callback_and_context");
+		return NULL;
+	}
+
+	Py_INCREF(callback_and_context);
+
+	if (msg_type == -1) {
+		uint32_t msg_type32 = msg_type;
+		status = imessaging_register_tmp(iface->msg_ctx, callback_and_context,
+						py_msg_callback_wrapper, &msg_type32);
+		msg_type = msg_type32;
+	} else {
+		status = imessaging_register(iface->msg_ctx, callback_and_context,
+				    msg_type, py_msg_callback_wrapper);
+	}
+	if (NT_STATUS_IS_ERR(status)) {
+		PyErr_SetNTSTATUS(status);
+		return NULL;
+	}
+
+	return PyLong_FromLong(msg_type);
+}
+
+static PyObject *py_imessaging_deregister(PyObject *self, PyObject *args, PyObject *kwargs)
+{
+	imessaging_Object *iface = (imessaging_Object *)self;
+	int msg_type = -1;
+	PyObject *callback;
+	const char *kwnames[] = { "callback", "msg_type", NULL };
+
+	if (!PyArg_ParseTupleAndKeywords(args, kwargs, "O|i:deregister",
+		discard_const_p(char *, kwnames), &callback, &msg_type)) {
+		return NULL;
+	}
+
+	imessaging_deregister(iface->msg_ctx, msg_type, callback);
+
+	Py_RETURN_NONE;
+}
+
+static void simple_timer_handler(struct tevent_context *ev,
+				 struct tevent_timer *te,
+				 struct timeval current_time,
+				 void *private_data)
+{
+	return;
+}
+
+static PyObject *py_imessaging_loop_once(PyObject *self, PyObject *args, PyObject *kwargs)
+{
+	imessaging_Object *iface = (imessaging_Object *)self;
+	double offset;
+	int seconds;
+	struct timeval next_event;
+	struct tevent_timer *timer = NULL;
+	const char *kwnames[] = { "timeout", NULL };
+
+	TALLOC_CTX *frame = talloc_stackframe();
+
+	if (!PyArg_ParseTupleAndKeywords(args, kwargs, "d",
+					 discard_const_p(char *, kwnames), &offset)) {
+		TALLOC_FREE(frame);
+		return NULL;
+	}
+
+	if (offset != 0.0) {
+		seconds = offset;
+		offset -= seconds;
+		next_event = tevent_timeval_current_ofs(seconds, (int)(offset*1000000));
+
+		timer = tevent_add_timer(iface->msg_ctx->ev, frame, next_event, simple_timer_handler,
+					 NULL);
+		if (timer == NULL) {
+			PyErr_NoMemory();
+			TALLOC_FREE(frame);
+			return NULL;
+		}
+	}
+
+	tevent_loop_once(iface->msg_ctx->ev);
+
+	TALLOC_FREE(frame);
+
+	Py_RETURN_NONE;
+}
+
+static PyObject *py_irpc_add_name(PyObject *self, PyObject *args)
+{
+	imessaging_Object *iface = (imessaging_Object *)self;
+	char *server_name;
+	NTSTATUS status;
+
+	if (!PyArg_ParseTuple(args, "s", &server_name)) {
+		return NULL;
+	}
+
+	status = irpc_add_name(iface->msg_ctx, server_name);
+	if (!NT_STATUS_IS_OK(status)) {
+		PyErr_SetNTSTATUS(status);
+		return NULL;
+	}
+
+	Py_RETURN_NONE;
+}
+
+static PyObject *py_irpc_remove_name(PyObject *self, PyObject *args)
+{
+	imessaging_Object *iface = (imessaging_Object *)self;
+	char *server_name;
+
+	if (!PyArg_ParseTuple(args, "s", &server_name)) {
+		return NULL;
+	}
+
+	irpc_remove_name(iface->msg_ctx, server_name);
+
+	Py_RETURN_NONE;
+}
+
+static PyObject *py_irpc_servers_byname(PyObject *self, PyObject *args)
+{
+	imessaging_Object *iface = (imessaging_Object *)self;
+	char *server_name;
+	unsigned i, num_ids;
+	struct server_id *ids;
+	PyObject *pylist;
+	TALLOC_CTX *mem_ctx = talloc_new(NULL);
+	NTSTATUS status;
+
+	if (!mem_ctx) {
+		PyErr_NoMemory();
+		return NULL;
+	}
+
+	if (!PyArg_ParseTuple(args, "s", &server_name)) {
+		TALLOC_FREE(mem_ctx);
+		return NULL;
+	}
+
+	status = irpc_servers_byname(iface->msg_ctx, mem_ctx, server_name,
+				     &num_ids, &ids);
+	if (!NT_STATUS_IS_OK(status)) {
+		TALLOC_FREE(mem_ctx);
+		PyErr_SetString(PyExc_KeyError, "No such name");
+		return NULL;
+	}
+
+	pylist = PyList_New(num_ids);
+	if (pylist == NULL) {
+		TALLOC_FREE(mem_ctx);
+		PyErr_NoMemory();
+		return NULL;
+	}
+	for (i = 0; i < num_ids; i++) {
+		PyObject *py_server_id;
+		struct server_id *p_server_id = talloc(NULL, struct server_id);
+		if (!p_server_id) {
+			PyErr_NoMemory();
+			return NULL;
+		}
+		*p_server_id = ids[i];
+
+		py_server_id = py_return_ndr_struct("samba.dcerpc.server_id", "server_id", p_server_id, p_server_id);
+		if (!py_server_id) {
+			return NULL;
+		}
+		PyList_SetItem(pylist, i, py_server_id);
+		talloc_unlink(NULL, p_server_id);
+	}
+	TALLOC_FREE(mem_ctx);
+	return pylist;
+}
+
+static PyObject *py_irpc_all_servers(PyObject *self,
+		PyObject *Py_UNUSED(ignored))
+{
+	imessaging_Object *iface = (imessaging_Object *)self;
+	PyObject *pylist;
+	int i;
+	struct irpc_name_records *records;
+	TALLOC_CTX *mem_ctx = talloc_new(NULL);
+	if (!mem_ctx) {
+		PyErr_NoMemory();
+		return NULL;
+	}
+
+	records = irpc_all_servers(iface->msg_ctx, mem_ctx);
+	if (records == NULL) {
+		return NULL;
+	}
+
+	pylist = PyList_New(records->num_records);
+	if (pylist == NULL) {
+		TALLOC_FREE(mem_ctx);
+		PyErr_NoMemory();
+		return NULL;
+	}
+	for (i = 0; i < records->num_records; i++) {
+		PyObject *py_name_record
+			= py_return_ndr_struct("samba.dcerpc.irpc",
+					       "name_record",
+					       records->names[i],
+					       records->names[i]);
+		if (!py_name_record) {
+			return NULL;
+		}
+		PyList_SetItem(pylist, i,
+			       py_name_record);
+	}
+	TALLOC_FREE(mem_ctx);
+	return pylist;
+}
+
+static PyMethodDef py_imessaging_methods[] = {
+	{ "send", PY_DISCARD_FUNC_SIG(PyCFunction, py_imessaging_send),
+		METH_VARARGS|METH_KEYWORDS,
+		"S.send(target, msg_type, data) -> None\nSend a message" },
+	{ "register", PY_DISCARD_FUNC_SIG(PyCFunction, py_imessaging_register),
+		METH_VARARGS|METH_KEYWORDS,
+		"S.register((callback, context), msg_type=None) -> msg_type\nRegister a message handler.  "
+	        "The callback and context must be supplied as a two-element tuple." },
+	{ "deregister", PY_DISCARD_FUNC_SIG(PyCFunction,
+					    py_imessaging_deregister),
+		METH_VARARGS|METH_KEYWORDS,
+		"S.deregister((callback, context), msg_type) -> None\nDeregister a message handler "
+	        "The callback and context must be supplied as the exact same two-element tuple "
+	        "as was used as registration time." },
+	{ "loop_once", PY_DISCARD_FUNC_SIG(PyCFunction,
+					   py_imessaging_loop_once),
+		METH_VARARGS|METH_KEYWORDS,
+		"S.loop_once(timeout) -> None\n"
+	        "Loop on the internal event context until we get an event "
+	        "(which might be a message calling the callback), "
+	        "timeout after timeout seconds (if not 0)" },
+	{ "irpc_add_name", (PyCFunction)py_irpc_add_name, METH_VARARGS,
+		"S.irpc_add_name(name) -> None\n"
+	        "Add this context to the list of server_id values that "
+	        "are registered for a particular name" },
+	{ "irpc_remove_name", (PyCFunction)py_irpc_remove_name, METH_VARARGS,
+		"S.irpc_remove_name(name) -> None\n"
+	        "Remove this context from the list of server_id values that "
+	        "are registered for a particular name" },
+	{ "irpc_servers_byname", (PyCFunction)py_irpc_servers_byname, METH_VARARGS,
+		"S.irpc_servers_byname(name) -> list\nGet list of server_id values that are registered for a particular name" },
+	{ "irpc_all_servers", (PyCFunction)py_irpc_all_servers, METH_NOARGS,
+		"S.irpc_all_servers() -> list\n"
+	        "Get list of all registered names and the associated server_id values" },
+	{ NULL, NULL, 0, NULL }
+};
+
+static PyObject *py_imessaging_server_id(PyObject *obj, void *closure)
+{
+	imessaging_Object *iface = (imessaging_Object *)obj;
+	PyObject *py_server_id;
+	struct server_id server_id = imessaging_get_server_id(iface->msg_ctx);
+	struct server_id *p_server_id = talloc(NULL, struct server_id);
+	if (!p_server_id) {
+		PyErr_NoMemory();
+		return NULL;
+	}
+	*p_server_id = server_id;
+
+	py_server_id = py_return_ndr_struct("samba.dcerpc.server_id", "server_id", p_server_id, p_server_id);
+	talloc_unlink(NULL, p_server_id);
+
+	return py_server_id;
+}
+
+static PyGetSetDef py_imessaging_getset[] = {
+	{
+		.name = discard_const_p(char, "server_id"),
+		.get  = py_imessaging_server_id,
+		.doc  = discard_const_p(char, "local server id")
+	},
+	{ .name = NULL },
+};
+
+
+PyTypeObject imessaging_Type = {
+	PyVarObject_HEAD_INIT(NULL, 0)
+	.tp_name = "messaging.Messaging",
+	.tp_basicsize = sizeof(imessaging_Object),
+	.tp_flags = Py_TPFLAGS_DEFAULT|Py_TPFLAGS_BASETYPE,
+	.tp_new = py_imessaging_connect,
+	.tp_dealloc = py_imessaging_dealloc,
+	.tp_methods = py_imessaging_methods,
+	.tp_getset = py_imessaging_getset,
+	.tp_doc = "Messaging(own_id=None)\n" \
+		  "Create a new object that can be used to communicate with the peers in the specified messaging path.\n"
+};
+
+static struct PyModuleDef moduledef = {
+    PyModuleDef_HEAD_INIT,
+    .m_name = "messaging",
+    .m_doc = "Internal RPC",
+    .m_size = -1,
+    .m_methods = NULL,
+};
+
+MODULE_INIT_FUNC(messaging)
+{
+	PyObject *mod;
+
+	if (PyType_Ready(&imessaging_Type) < 0)
+		return NULL;
+
+	mod = PyModule_Create(&moduledef);
+	if (mod == NULL)
+		return NULL;
+
+	Py_INCREF((PyObject *)&imessaging_Type);
+	PyModule_AddObject(mod, "Messaging", (PyObject *)&imessaging_Type);
+	PyModule_AddObject(mod, "IRPC_CALL_TIMEOUT", PyLong_FromLong(IRPC_CALL_TIMEOUT));
+	PyModule_AddObject(mod, "IRPC_CALL_TIMEOUT_INF", PyLong_FromLong(IRPC_CALL_TIMEOUT_INF));
+
+	return mod;
+}
diff --git a/source4/lib/messaging/tests/irpc.c b/source4/lib/messaging/tests/irpc.c
new file mode 100644
index 0000000..466b47f
--- /dev/null
+++ b/source4/lib/messaging/tests/irpc.c
@@ -0,0 +1,308 @@
+/* 
+   Unix SMB/CIFS implementation.
+
+   local test for irpc code
+
+   Copyright (C) Andrew Tridgell 2004
+   
+   This program is free software; you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation; either version 3 of the License, or
+   (at your option) any later version.
+   
+   This program 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 General Public License for more details.
+   
+   You should have received a copy of the GNU General Public License
+   along with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#include "includes.h"
+#include "lib/events/events.h"
+#include "lib/messaging/irpc.h"
+#include "librpc/gen_ndr/ndr_echo.h"
+#include "librpc/gen_ndr/ndr_echo_c.h"
+#include "torture/torture.h"
+#include "cluster/cluster.h"
+#include "param/param.h"
+#include "torture/local/proto.h"
+
+const uint32_t MSG_ID1 = 1, MSG_ID2 = 2;
+
+static bool test_debug;
+
+struct irpc_test_data
+{
+	struct imessaging_context *msg_ctx1, *msg_ctx2;
+	struct tevent_context *ev;
+};
+
+/*
+  serve up AddOne over the irpc system
+*/
+static NTSTATUS irpc_AddOne(struct irpc_message *irpc, struct echo_AddOne *r)
+{
+	*r->out.out_data = r->in.in_data + 1;
+	if (test_debug) {
+		printf("irpc_AddOne: in=%u in+1=%u out=%u\n", 
+			r->in.in_data, r->in.in_data+1, *r->out.out_data);
+	}
+	return NT_STATUS_OK;
+}
+
+/*
+  a deferred reply to echodata
+*/
+static void deferred_echodata(struct tevent_context *ev, struct tevent_timer *te, 
+			      struct timeval t, void *private_data)
+{
+	struct irpc_message *irpc = talloc_get_type(private_data, struct irpc_message);
+	struct echo_EchoData *r = (struct echo_EchoData *)irpc->data;
+	r->out.out_data = (uint8_t *)talloc_memdup(r, r->in.in_data, r->in.len);
+	if (r->out.out_data == NULL) {
+		irpc_send_reply(irpc, NT_STATUS_NO_MEMORY);
+	}
+	printf("sending deferred reply\n");
+	irpc_send_reply(irpc, NT_STATUS_OK);
+}
+
+
+/*
+  serve up EchoData over the irpc system
+*/
+static NTSTATUS irpc_EchoData(struct irpc_message *irpc, struct echo_EchoData *r)
+{
+	struct irpc_test_data *data = talloc_get_type_abort(irpc->private_data, struct irpc_test_data);
+	irpc->defer_reply = true;
+	tevent_add_timer(data->ev, irpc, timeval_zero(), deferred_echodata, irpc);
+	return NT_STATUS_OK;
+}
+
+
+/*
+  test a addone call over the internal messaging system
+*/
+static bool test_addone(struct torture_context *test, const void *_data,
+			const void *_value)
+{
+	struct echo_AddOne r;
+	NTSTATUS status;
+	const struct irpc_test_data *data = (const struct irpc_test_data *)_data;
+	uint32_t value = *(const uint32_t *)_value;
+	struct dcerpc_binding_handle *irpc_handle;
+
+	irpc_handle = irpc_binding_handle(test, data->msg_ctx1,
+					  cluster_id(0, MSG_ID2),
+					  &ndr_table_rpcecho);
+	torture_assert(test, irpc_handle, "no memory");
+
+	/* make the call */
+	r.in.in_data = value;
+
+	test_debug = true;
+	/*
+	 * Note: this makes use of nested event loops
+	 * as client and server use the same loop.
+	 */
+	dcerpc_binding_handle_set_sync_ev(irpc_handle, data->ev);
+	status = dcerpc_echo_AddOne_r(irpc_handle, test, &r);
+	test_debug = false;
+	torture_assert_ntstatus_ok(test, status, "AddOne failed");
+
+	/* check the answer */
+	torture_assert(test, *r.out.out_data == r.in.in_data + 1, 
+				   "AddOne wrong answer");
+
+	torture_comment(test, "%u + 1 = %u\n", r.in.in_data, *r.out.out_data);
+	return true;
+}
+
+/*
+  test a echodata call over the internal messaging system
+*/
+static bool test_echodata(struct torture_context *tctx,
+						  const void *tcase_data,
+						  const void *test_data)
+{
+	struct echo_EchoData r;
+	NTSTATUS status;
+	const struct irpc_test_data *data = (const struct irpc_test_data *)tcase_data;
+	TALLOC_CTX *mem_ctx = tctx;
+	struct dcerpc_binding_handle *irpc_handle;
+
+	irpc_handle = irpc_binding_handle(mem_ctx, data->msg_ctx1,
+					  cluster_id(0, MSG_ID2),
+					  &ndr_table_rpcecho);
+	torture_assert(tctx, irpc_handle, "no memory");
+
+	/* make the call */
+	r.in.in_data = (unsigned char *)talloc_strdup(mem_ctx, "0123456789");
+	r.in.len = strlen((char *)r.in.in_data);
+
+	/*
+	 * Note: this makes use of nested event loops
+	 * as client and server use the same loop.
+	 */
+	dcerpc_binding_handle_set_sync_ev(irpc_handle, data->ev);
+	status = dcerpc_echo_EchoData_r(irpc_handle, mem_ctx, &r);
+	torture_assert_ntstatus_ok(tctx, status, "EchoData failed");
+
+	/* check the answer */
+	if (memcmp(r.out.out_data, r.in.in_data, r.in.len) != 0) {
+		NDR_PRINT_OUT_DEBUG(echo_EchoData, &r);
+		torture_fail(tctx, "EchoData wrong answer");
+	}
+
+	torture_comment(tctx, "Echo '%*.*s' -> '%*.*s'\n", 
+	       r.in.len, r.in.len,
+	       r.in.in_data,
+	       r.in.len, r.in.len,
+	       r.out.out_data);
+	return true;
+}
+
+struct irpc_callback_state {
+	struct echo_AddOne r;
+	int *pong_count;
+};
+
+static void irpc_callback(struct tevent_req *subreq)
+{
+	struct irpc_callback_state *s =
+		tevent_req_callback_data(subreq,
+		struct irpc_callback_state);
+	NTSTATUS status;
+
+	status = dcerpc_echo_AddOne_r_recv(subreq, s);
+	TALLOC_FREE(subreq);
+	if (!NT_STATUS_IS_OK(status)) {
+		printf("irpc call failed - %s\n", nt_errstr(status));
+	}
+	if (*s->r.out.out_data != s->r.in.in_data + 1) {
+		printf("AddOne wrong answer - %u + 1 = %u should be %u\n", 
+		       s->r.in.in_data, *s->r.out.out_data, s->r.in.in_data+1);
+	}
+	(*s->pong_count)++;
+}
+
+/*
+  test echo speed
+*/
+static bool test_speed(struct torture_context *tctx,
+					   const void *tcase_data,
+					   const void *test_data)
+{
+	int ping_count = 0;
+	int pong_count = 0;
+	const struct irpc_test_data *data = (const struct irpc_test_data *)tcase_data;
+	struct timeval tv;
+	TALLOC_CTX *mem_ctx = tctx;
+	int timelimit = torture_setting_int(tctx, "timelimit", 10);
+	struct dcerpc_binding_handle *irpc_handle;
+
+	irpc_handle = irpc_binding_handle(mem_ctx, data->msg_ctx1,
+					  cluster_id(0, MSG_ID2),
+					  &ndr_table_rpcecho);
+	torture_assert(tctx, irpc_handle, "no memory");
+
+	tv = timeval_current();
+
+	torture_comment(tctx, "Sending echo for %d seconds\n", timelimit);
+	while (timeval_elapsed(&tv) < timelimit) {
+		struct tevent_req *subreq;
+		struct irpc_callback_state *s;
+
+		s = talloc_zero(mem_ctx, struct irpc_callback_state);
+		torture_assert(tctx, s != NULL, "no mem");
+
+		s->pong_count = &pong_count;
+
+		subreq = dcerpc_echo_AddOne_r_send(mem_ctx,
+						   tctx->ev,
+						   irpc_handle,
+						   &s->r);
+		torture_assert(tctx, subreq != NULL, "AddOne send failed");
+
+		tevent_req_set_callback(subreq, irpc_callback, s);
+
+		ping_count++;
+
+		while (ping_count > pong_count + 20) {
+			tevent_loop_once(data->ev);
+		}
+	}
+
+	torture_comment(tctx, "waiting for %d remaining replies (done %d)\n", 
+	       ping_count - pong_count, pong_count);
+	while (timeval_elapsed(&tv) < 30 && pong_count < ping_count) {
+		tevent_loop_once(data->ev);
+	}
+
+	torture_assert_int_equal(tctx, ping_count, pong_count, "ping test failed");
+
+	torture_comment(tctx, "echo rate of %.0f messages/sec\n", 
+	       (ping_count+pong_count)/timeval_elapsed(&tv));
+	return true;
+}
+
+
+static bool irpc_setup(struct torture_context *tctx, void **_data)
+{
+	struct irpc_test_data *data;
+
+	*_data = data = talloc(tctx, struct irpc_test_data);
+
+	lpcfg_set_cmdline(tctx->lp_ctx, "pid directory", "piddir.tmp");
+
+	data->ev = tctx->ev;
+	torture_assert(tctx, data->msg_ctx1 = 
+		       imessaging_init(tctx,
+				      tctx->lp_ctx,
+				      cluster_id(0, MSG_ID1),
+				      data->ev),
+		       "Failed to init first messaging context");
+
+	torture_assert(tctx, data->msg_ctx2 = 
+		       imessaging_init(tctx,
+				      tctx->lp_ctx,
+				      cluster_id(0, MSG_ID2), 
+				      data->ev),
+		       "Failed to init second messaging context");
+
+	/* register the server side function */
+	IRPC_REGISTER(data->msg_ctx1, rpcecho, ECHO_ADDONE, irpc_AddOne, data);
+	IRPC_REGISTER(data->msg_ctx2, rpcecho, ECHO_ADDONE, irpc_AddOne, data);
+
+	IRPC_REGISTER(data->msg_ctx1, rpcecho, ECHO_ECHODATA, irpc_EchoData, data);
+	IRPC_REGISTER(data->msg_ctx2, rpcecho, ECHO_ECHODATA, irpc_EchoData, data);
+
+	return true;
+}
+
+struct torture_suite *torture_local_irpc(TALLOC_CTX *mem_ctx)
+{
+	struct torture_suite *suite = torture_suite_create(mem_ctx, "irpc");
+	struct torture_tcase *tcase = torture_suite_add_tcase(suite, "irpc");
+	int i;
+	uint32_t *values = talloc_array(tcase, uint32_t, 5);
+
+	values[0] = 0;
+	values[1] = 0x7FFFFFFE;
+	values[2] = 0xFFFFFFFE;
+	values[3] = 0xFFFFFFFF;
+	values[4] = random() & 0xFFFFFFFF;
+
+	tcase->setup = irpc_setup;
+
+	for (i = 0; i < 5; i++) {
+		torture_tcase_add_test_const(tcase, "addone", test_addone,
+				(void *)&values[i]);
+	}
+
+	torture_tcase_add_test_const(tcase, "echodata", test_echodata, NULL);
+	torture_tcase_add_test_const(tcase, "speed", test_speed, NULL);
+
+	return suite;
+}
diff --git a/source4/lib/messaging/tests/messaging.c b/source4/lib/messaging/tests/messaging.c
new file mode 100644
index 0000000..dcbbc19
--- /dev/null
+++ b/source4/lib/messaging/tests/messaging.c
@@ -0,0 +1,694 @@
+/* 
+   Unix SMB/CIFS implementation.
+
+   local test for messaging code
+
+   Copyright (C) Andrew Tridgell 2004
+   
+   This program is free software; you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation; either version 3 of the License, or
+   (at your option) any later version.
+   
+   This program 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 General Public License for more details.
+   
+   You should have received a copy of the GNU General Public License
+   along with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#include "includes.h"
+#include "lib/events/events.h"
+#include "lib/messaging/irpc.h"
+#include "torture/torture.h"
+#include "cluster/cluster.h"
+#include "param/param.h"
+#include "torture/local/proto.h"
+#include "system/select.h"
+#include "system/filesys.h"
+
+#include <gnutls/gnutls.h>
+#include <gnutls/crypto.h>
+
+static uint32_t msg_pong;
+
+static void ping_message(struct imessaging_context *msg,
+			 void *private_data,
+			 uint32_t msg_type,
+			 struct server_id src,
+			 size_t num_fds,
+			 int *fds,
+			 DATA_BLOB *data)
+{
+	NTSTATUS status;
+
+	if (num_fds != 0) {
+		DBG_WARNING("Received %zu fds, ignoring message\n", num_fds);
+		return;
+	}
+
+	status = imessaging_send(msg, src, msg_pong, data);
+	if (!NT_STATUS_IS_OK(status)) {
+		printf("pong failed - %s\n", nt_errstr(status));
+	}
+}
+
+static void pong_message(struct imessaging_context *msg,
+			 void *private_data,
+			 uint32_t msg_type,
+			 struct server_id src,
+			 size_t num_fds,
+			 int *fds,
+			 DATA_BLOB *data)
+{
+	int *count = (int *)private_data;
+
+	if (num_fds != 0) {
+		DBG_WARNING("Received %zu fds, ignoring message\n", num_fds);
+		return;
+	}
+
+	(*count)++;
+}
+
+static void exit_message(struct imessaging_context *msg,
+			 void *private_data,
+			 uint32_t msg_type,
+			 struct server_id src,
+			 size_t num_fds,
+			 int *fds,
+			 DATA_BLOB *data)
+{
+	if (num_fds != 0) {
+		DBG_WARNING("Received %zu fds, ignoring message\n", num_fds);
+		return;
+	}
+
+	talloc_free(private_data);
+	exit(0);
+}
+
+/*
+  test ping speed
+*/
+static bool test_ping_speed(struct torture_context *tctx)
+{
+	struct tevent_context *ev;
+	struct imessaging_context *msg_client_ctx;
+	struct imessaging_context *msg_server_ctx;
+	int ping_count = 0;
+	int pong_count = 0;
+	struct timeval tv;
+	int timelimit = torture_setting_int(tctx, "timelimit", 10);
+	uint32_t msg_ping, msg_exit;
+
+	lpcfg_set_cmdline(tctx->lp_ctx, "pid directory", "piddir.tmp");
+
+	ev = tctx->ev;
+
+	msg_server_ctx = imessaging_init(tctx,
+					 tctx->lp_ctx, cluster_id(0, 1),
+					 ev);
+	
+	torture_assert(tctx, msg_server_ctx != NULL, "Failed to init ping messaging context");
+		
+	imessaging_register_tmp(msg_server_ctx, NULL, ping_message, &msg_ping);
+	imessaging_register_tmp(msg_server_ctx, tctx, exit_message, &msg_exit);
+
+	msg_client_ctx = imessaging_init(tctx,
+					 tctx->lp_ctx,
+					 cluster_id(0, 2),
+					 ev);
+
+	torture_assert(tctx, msg_client_ctx != NULL, 
+		       "msg_client_ctx imessaging_init() failed");
+
+	imessaging_register_tmp(msg_client_ctx, &pong_count, pong_message, &msg_pong);
+
+	tv = timeval_current();
+
+	torture_comment(tctx, "Sending pings for %d seconds\n", timelimit);
+	while (timeval_elapsed(&tv) < timelimit) {
+		DATA_BLOB data;
+		NTSTATUS status1, status2;
+
+		data.data = discard_const_p(uint8_t, "testing");
+		data.length = strlen((const char *)data.data);
+
+		status1 = imessaging_send(msg_client_ctx, cluster_id(0, 1), msg_ping, &data);
+		status2 = imessaging_send(msg_client_ctx, cluster_id(0, 1), msg_ping, NULL);
+
+		torture_assert_ntstatus_ok(tctx, status1, "msg1 failed");
+		ping_count++;
+
+		torture_assert_ntstatus_ok(tctx, status2, "msg2 failed");
+		ping_count++;
+
+		while (ping_count > pong_count + 20) {
+			tevent_loop_once(ev);
+		}
+	}
+
+	torture_comment(tctx, "waiting for %d remaining replies (done %d)\n", 
+	       ping_count - pong_count, pong_count);
+	while (timeval_elapsed(&tv) < 30 && pong_count < ping_count) {
+		tevent_loop_once(ev);
+	}
+
+	torture_comment(tctx, "sending exit\n");
+	imessaging_send(msg_client_ctx, cluster_id(0, 1), msg_exit, NULL);
+
+	torture_assert_int_equal(tctx, ping_count, pong_count, "ping test failed");
+
+	torture_comment(tctx, "ping rate of %.0f messages/sec\n", 
+	       (ping_count+pong_count)/timeval_elapsed(&tv));
+
+	talloc_free(msg_client_ctx);
+	talloc_free(msg_server_ctx);
+
+	return true;
+}
+
+static bool test_messaging_overflow(struct torture_context *tctx)
+{
+	struct imessaging_context *msg_ctx;
+	ssize_t nwritten, nread;
+	pid_t child;
+	char c = 0;
+	int up_pipe[2], down_pipe[2];
+	int i, ret, child_status;
+
+	ret = pipe(up_pipe);
+	torture_assert(tctx, ret == 0, "pipe failed");
+	ret = pipe(down_pipe);
+	torture_assert(tctx, ret == 0, "pipe failed");
+
+	child = fork();
+	if (child < 0) {
+		torture_fail(tctx, "fork failed");
+	}
+
+	if (child == 0) {
+		ret = tevent_re_initialise(tctx->ev);
+		torture_assert(tctx, ret == 0, "tevent_re_initialise failed");
+
+		msg_ctx = imessaging_init(tctx, tctx->lp_ctx,
+					  cluster_id(getpid(), 0),
+					  tctx->ev);
+		torture_assert(tctx, msg_ctx != NULL,
+			       "imessaging_init failed");
+
+		do {
+			nwritten = write(up_pipe[1], &c, 1);
+		} while ((nwritten == -1) && (errno == EINTR));
+
+		ret = close(down_pipe[1]);
+		torture_assert(tctx, ret == 0, "close failed");
+
+		do {
+			nread = read(down_pipe[0], &c, 1);
+		} while ((nread == -1) && (errno == EINTR));
+
+		exit(0);
+	}
+
+	do {
+		nread = read(up_pipe[0], &c, 1);
+	} while ((nread == -1) && (errno == EINTR));
+
+	msg_ctx = imessaging_init(tctx, tctx->lp_ctx, cluster_id(getpid(), 0),
+				  tctx->ev);
+	torture_assert(tctx, msg_ctx != NULL, "imessaging_init failed");
+
+	for (i=0; i<1000; i++) {
+		NTSTATUS status;
+		status = imessaging_send(msg_ctx, cluster_id(child, 0),
+					 MSG_PING, NULL);
+		torture_assert_ntstatus_ok(tctx, status,
+					   "imessaging_send failed");
+	}
+
+	tevent_loop_once(tctx->ev);
+
+	talloc_free(msg_ctx);
+
+	ret = close(down_pipe[1]);
+	torture_assert(tctx, ret == 0, "close failed");
+
+	ret = waitpid(child, &child_status, 0);
+	torture_assert(tctx, ret == child, "wrong child exited");
+	torture_assert(tctx, child_status == 0, "child failed");
+
+	poll(NULL, 0, 500);
+
+	return true;
+}
+
+struct overflow_parent_child {
+	gnutls_hash_hd_t md5_hash_hnd;
+	bool done;
+};
+
+static void overflow_md5_child_handler(struct imessaging_context *msg,
+				       void *private_data,
+				       uint32_t msg_type,
+				       struct server_id server_id,
+				       size_t num_fds,
+				       int *fds,
+				       DATA_BLOB *data)
+{
+	struct overflow_parent_child *state = private_data;
+
+	if (num_fds != 0) {
+		DBG_WARNING("Received %zu fds, ignoring message\n", num_fds);
+		return;
+	}
+
+	if (data->length == 0) {
+		state->done = true;
+		return;
+	}
+
+	gnutls_hash(state->md5_hash_hnd, data->data, data->length);
+}
+
+struct overflow_child_parent {
+	uint8_t final[16];
+	bool done;
+};
+
+static void overflow_md5_parent_handler(struct imessaging_context *msg_ctx,
+					void *private_data,
+					uint32_t msg_type,
+					struct server_id server_id,
+					size_t num_fds,
+					int *fds,
+					DATA_BLOB *data)
+{
+	struct overflow_child_parent *state = private_data;
+
+	if (num_fds != 0) {
+		DBG_WARNING("Received %zu fds, ignoring message\n", num_fds);
+		return;
+	}
+
+	if (data->length != sizeof(state->final)) {
+		memset(state->final, 0, sizeof(state->final));
+		state->done = true;
+		return;
+	}
+	memcpy(state->final, data->data, 16);
+	state->done = true;
+}
+
+static bool test_messaging_overflow_check(struct torture_context *tctx)
+{
+	struct imessaging_context *msg_ctx;
+	ssize_t nwritten, nread;
+	pid_t child;
+	char c = 0;
+	int up_pipe[2], down_pipe[2];
+	int i, ret, child_status;
+	gnutls_hash_hd_t hash_hnd;
+	uint8_t final[16];
+	struct overflow_child_parent child_msg = { .done = false };
+	NTSTATUS status;
+
+	ret = pipe(up_pipe);
+	torture_assert(tctx, ret == 0, "pipe failed");
+	ret = pipe(down_pipe);
+	torture_assert(tctx, ret == 0, "pipe failed");
+
+	child = fork();
+	if (child < 0) {
+		torture_fail(tctx, "fork failed");
+	}
+
+	if (child == 0) {
+		struct overflow_parent_child child_state = { .done = false };
+		DATA_BLOB retblob = { .data = final, .length = sizeof(final) };
+
+		ret = tevent_re_initialise(tctx->ev);
+		torture_assert(tctx, ret == 0, "tevent_re_initialise failed");
+
+		gnutls_hash_init(&child_state.md5_hash_hnd, GNUTLS_DIG_MD5);
+
+		msg_ctx = imessaging_init(tctx, tctx->lp_ctx,
+					  cluster_id(getpid(), 0),
+					  tctx->ev);
+		torture_assert(tctx, msg_ctx != NULL,
+			       "imessaging_init failed");
+
+		status = imessaging_register(msg_ctx, &child_state,
+					     MSG_TMP_BASE-1,
+					     overflow_md5_child_handler);
+		torture_assert(tctx, NT_STATUS_IS_OK(status),
+			       "imessaging_register failed");
+
+		do {
+			nwritten = write(up_pipe[1], &c, 1);
+		} while ((nwritten == -1) && (errno == EINTR));
+
+		ret = close(down_pipe[1]);
+		torture_assert(tctx, ret == 0, "close failed");
+
+		do {
+			nread = read(down_pipe[0], &c, 1);
+		} while ((nread == -1) && (errno == EINTR));
+
+		while (!child_state.done) {
+			tevent_loop_once(tctx->ev);
+		}
+
+		gnutls_hash_deinit(child_state.md5_hash_hnd, final);
+
+		status = imessaging_send(msg_ctx,
+					 cluster_id(getppid(), 0),
+					 MSG_TMP_BASE-2,
+					 &retblob);
+		torture_assert(tctx, NT_STATUS_IS_OK(status),
+			       "imessaging_send failed");
+
+		exit(0);
+	}
+
+	do {
+		nread = read(up_pipe[0], &c, 1);
+	} while ((nread == -1) && (errno == EINTR));
+
+	msg_ctx = imessaging_init(tctx, tctx->lp_ctx, cluster_id(getpid(), 0),
+				  tctx->ev);
+	torture_assert(tctx, msg_ctx != NULL, "imessaging_init failed");
+
+	status = imessaging_register(msg_ctx,
+				     &child_msg,
+				     MSG_TMP_BASE-2,
+				     overflow_md5_parent_handler);
+	torture_assert(tctx,
+		       NT_STATUS_IS_OK(status),
+		       "imessaging_register failed");
+
+	gnutls_hash_init(&hash_hnd, GNUTLS_DIG_MD5);
+
+	for (i=0; i<1000; i++) {
+		size_t len = ((random() % 100) + 1);
+		uint8_t buf[len];
+		DATA_BLOB blob = { .data = buf, .length = len };
+
+		generate_random_buffer(buf, len);
+
+		gnutls_hash(hash_hnd, buf, len);
+
+		status = imessaging_send(msg_ctx, cluster_id(child, 0),
+					 MSG_TMP_BASE-1, &blob);
+		torture_assert_ntstatus_ok(tctx, status,
+					   "imessaging_send failed");
+	}
+
+	status = imessaging_send(msg_ctx, cluster_id(child, 0),
+				 MSG_TMP_BASE-1, NULL);
+	torture_assert_ntstatus_ok(tctx, status,
+				   "imessaging_send failed");
+
+	gnutls_hash_deinit(hash_hnd, final);
+
+	do {
+		nwritten = write(down_pipe[1], &c, 1);
+	} while ((nwritten == -1) && (errno == EINTR));
+
+	while (!child_msg.done) {
+		tevent_loop_once(tctx->ev);
+	}
+
+	ret = close(down_pipe[1]);
+	torture_assert(tctx, ret == 0, "close failed");
+
+	talloc_free(msg_ctx);
+
+	ret = waitpid(child, &child_status, 0);
+	torture_assert(tctx, ret == child, "wrong child exited");
+	torture_assert(tctx, child_status == 0, "child failed");
+
+	if (memcmp(final, child_msg.final, 16) != 0) {
+		dump_data_file(final, 16, false, stderr);
+		dump_data_file(child_msg.final, 16, false, stderr);
+		fflush(stderr);
+		torture_fail(tctx, "checksum comparison failed");
+	}
+
+	return true;
+}
+
+struct test_multi_ctx {
+	struct torture_context *tctx;
+	struct imessaging_context *server_ctx;
+	struct imessaging_context *client_ctx[4];
+	size_t num_missing;
+	bool got_server;
+	bool got_client_0_1;
+	bool got_client_2_3;
+	bool ok;
+};
+
+static void multi_ctx_server_handler(struct imessaging_context *msg,
+				     void *private_data,
+				     uint32_t msg_type,
+				     struct server_id server_id,
+				     size_t num_fds,
+				     int *fds,
+				     DATA_BLOB *data)
+{
+	struct test_multi_ctx *state = private_data;
+	char *str = NULL;
+
+	if (num_fds != 0) {
+		DBG_WARNING("Received %zu fds, ignoring message\n", num_fds);
+		return;
+	}
+
+	torture_assert_goto(state->tctx, state->num_missing >= 1,
+			    state->ok, fail,
+			    "num_missing should be at least 1.");
+	state->num_missing -= 1;
+
+	torture_assert_goto(state->tctx, !state->got_server,
+			    state->ok, fail,
+			    "already got server.");
+	state->got_server = true;
+
+	/*
+	 * We free the context itself and most likely reuse
+	 * the memory immediately.
+	 */
+	TALLOC_FREE(state->server_ctx);
+	str = generate_random_str(state->tctx, 128);
+	torture_assert_goto(state->tctx, str != NULL,
+			    state->ok, fail,
+			    "generate_random_str()");
+
+fail:
+	return;
+}
+
+static void multi_ctx_client_0_1_handler(struct imessaging_context *msg,
+					 void *private_data,
+					 uint32_t msg_type,
+					 struct server_id server_id,
+					 size_t num_fds,
+					 int *fds,
+					 DATA_BLOB *data)
+{
+	struct test_multi_ctx *state = private_data;
+	char *str = NULL;
+
+	if (num_fds != 0) {
+		DBG_WARNING("Received %zu fds, ignoring message\n", num_fds);
+		return;
+	}
+
+	torture_assert_goto(state->tctx, state->num_missing >= 2,
+			    state->ok, fail,
+			    "num_missing should be at least 2.");
+	state->num_missing -= 2;
+
+	torture_assert_goto(state->tctx, !state->got_client_0_1,
+			    state->ok, fail,
+			    "already got client_0_1.");
+	state->got_client_0_1 = true;
+
+	/*
+	 * We free two contexts and most likely reuse
+	 * the memory immediately.
+	 */
+	TALLOC_FREE(state->client_ctx[0]);
+	str = generate_random_str(state->tctx, 128);
+	torture_assert_goto(state->tctx, str != NULL,
+			    state->ok, fail,
+			    "generate_random_str()");
+	TALLOC_FREE(state->client_ctx[1]);
+	str = generate_random_str(state->tctx, 128);
+	torture_assert_goto(state->tctx, str != NULL,
+			    state->ok, fail,
+			    "generate_random_str()");
+
+fail:
+	return;
+}
+
+static void multi_ctx_client_2_3_handler(struct imessaging_context *msg,
+					 void *private_data,
+					 uint32_t msg_type,
+					 struct server_id server_id,
+					 size_t num_fds,
+					 int *fds,
+					 DATA_BLOB *data)
+{
+	struct test_multi_ctx *state = private_data;
+	char *str = NULL;
+
+	if (num_fds != 0) {
+		DBG_WARNING("Received %zu fds, ignoring message\n", num_fds);
+		return;
+	}
+
+	torture_assert_goto(state->tctx, state->num_missing >= 2,
+			    state->ok, fail,
+			    "num_missing should be at least 2.");
+	state->num_missing -= 2;
+
+	torture_assert_goto(state->tctx, !state->got_client_2_3,
+			    state->ok, fail,
+			    "already got client_2_3.");
+	state->got_client_2_3 = true;
+
+	/*
+	 * We free two contexts and most likely reuse
+	 * the memory immediately.
+	 */
+	TALLOC_FREE(state->client_ctx[2]);
+	str = generate_random_str(state->tctx, 128);
+	torture_assert_goto(state->tctx, str != NULL,
+			    state->ok, fail,
+			    "generate_random_str()");
+	TALLOC_FREE(state->client_ctx[3]);
+	str = generate_random_str(state->tctx, 128);
+	torture_assert_goto(state->tctx, str != NULL,
+			    state->ok, fail,
+			    "generate_random_str()");
+
+fail:
+	return;
+}
+
+static bool test_multi_ctx(struct torture_context *tctx)
+{
+	struct test_multi_ctx state = {
+		.tctx = tctx,
+		.ok = true,
+	};
+	struct timeval tv;
+	NTSTATUS status;
+
+	lpcfg_set_cmdline(tctx->lp_ctx, "pid directory", "piddir.tmp");
+
+	/*
+	 * We use cluster_id(0, 0) as that gets for
+	 * all task ids.
+	 */
+	state.server_ctx = imessaging_init(tctx,
+					   tctx->lp_ctx,
+					   cluster_id(0, 0),
+					   tctx->ev);
+	torture_assert(tctx, state.server_ctx != NULL,
+		       "Failed to init messaging context");
+
+	status = imessaging_register(state.server_ctx, &state,
+				     MSG_TMP_BASE-1,
+				     multi_ctx_server_handler);
+	torture_assert(tctx, NT_STATUS_IS_OK(status), "imessaging_register failed");
+
+	state.client_ctx[0] = imessaging_init(tctx,
+					      tctx->lp_ctx,
+					      cluster_id(0, 0),
+					      tctx->ev);
+	torture_assert(tctx, state.client_ctx[0] != NULL,
+		       "msg_client_ctx imessaging_init() failed");
+	status = imessaging_register(state.client_ctx[0], &state,
+				     MSG_TMP_BASE-1,
+				     multi_ctx_client_0_1_handler);
+	torture_assert(tctx, NT_STATUS_IS_OK(status), "imessaging_register failed");
+	state.client_ctx[1] = imessaging_init(tctx,
+					      tctx->lp_ctx,
+					      cluster_id(0, 0),
+					      tctx->ev);
+	torture_assert(tctx, state.client_ctx[1] != NULL,
+		       "msg_client_ctx imessaging_init() failed");
+	status = imessaging_register(state.client_ctx[1], &state,
+				     MSG_TMP_BASE-1,
+				     multi_ctx_client_0_1_handler);
+	torture_assert(tctx, NT_STATUS_IS_OK(status), "imessaging_register failed");
+	state.client_ctx[2] = imessaging_init(tctx,
+					      tctx->lp_ctx,
+					      cluster_id(0, 0),
+					      tctx->ev);
+	torture_assert(tctx, state.client_ctx[2] != NULL,
+		       "msg_client_ctx imessaging_init() failed");
+	status = imessaging_register(state.client_ctx[2], &state,
+				     MSG_TMP_BASE-1,
+				     multi_ctx_client_2_3_handler);
+	torture_assert(tctx, NT_STATUS_IS_OK(status), "imessaging_register failed");
+	state.client_ctx[3] = imessaging_init(tctx,
+					      tctx->lp_ctx,
+					      cluster_id(0, 0),
+					      tctx->ev);
+	torture_assert(tctx, state.client_ctx[3] != NULL,
+		       "msg_client_ctx imessaging_init() failed");
+	status = imessaging_register(state.client_ctx[3], &state,
+				     MSG_TMP_BASE-1,
+				     multi_ctx_client_2_3_handler);
+	torture_assert(tctx, NT_STATUS_IS_OK(status), "imessaging_register failed");
+
+	/*
+	 * Send one message that need to arrive on 3 ( 5 - 2 ) handlers.
+	 */
+	state.num_missing = 5;
+
+	status = imessaging_send(state.server_ctx,
+				 cluster_id(0, 0),
+				 MSG_TMP_BASE-1, NULL);
+	torture_assert_ntstatus_ok(tctx, status, "msg failed");
+
+	tv = timeval_current();
+	while (timeval_elapsed(&tv) < 30 && state.num_missing > 0 && state.ok) {
+		int ret;
+
+		ret = tevent_loop_once(tctx->ev);
+		torture_assert_int_equal(tctx, ret, 0, "tevent_loop_once()");
+	}
+
+	if (!state.ok) {
+		return false;
+	}
+
+	torture_assert_int_equal(tctx, state.num_missing, 0,
+				 "wrong message count");
+
+	torture_assert(tctx, state.got_client_0_1, "got_client_0_1");
+	torture_assert(tctx, state.got_client_2_3, "got_client_2_3");
+	torture_assert(tctx, state.got_server, "got_server");
+
+	return true;
+}
+
+struct torture_suite *torture_local_messaging(TALLOC_CTX *mem_ctx)
+{
+	struct torture_suite *s = torture_suite_create(mem_ctx, "messaging");
+	torture_suite_add_simple_test(s, "overflow", test_messaging_overflow);
+	torture_suite_add_simple_test(s, "overflow_check",
+				      test_messaging_overflow_check);
+	torture_suite_add_simple_test(s, "ping_speed", test_ping_speed);
+	torture_suite_add_simple_test(s, "multi_ctx", test_multi_ctx);
+	return s;
+}
diff --git a/source4/lib/messaging/wscript_build b/source4/lib/messaging/wscript_build
new file mode 100644
index 0000000..3408396
--- /dev/null
+++ b/source4/lib/messaging/wscript_build
@@ -0,0 +1,33 @@
+#!/usr/bin/env python
+
+
+bld.SAMBA_LIBRARY('MESSAGING_SEND',
+	source='messaging_send.c',
+	public_deps='messages_util messages_dgm UNIX_PRIVS cluster server_id_db',
+	private_library=True
+	)
+
+bld.SAMBA_LIBRARY('MESSAGING',
+	source='messaging.c messaging_handlers.c',
+	public_deps='''
+            samba-util
+            NDR_IRPC
+            UNIX_PRIVS
+            cluster
+            ndr
+            dcerpc
+            messages_util
+            server_id_db
+            talloc_report_printf
+            ''',
+	private_library=True
+	)
+
+pyparam_util = bld.pyembed_libname('pyparam_util')
+pytalloc_util = bld.pyembed_libname('pytalloc-util')
+
+bld.SAMBA_PYTHON('python_messaging',
+		source='pymessaging.c',
+		deps='MESSAGING events %s %s' % (pyparam_util, pytalloc_util),
+		realname='samba/messaging.so'
+		)