--- a/configure.ac	Fri Nov 22 19:32:06 2013 +0200
+++ b/configure.ac	Fri Nov 22 19:39:13 2013 +0200
@@ -2884,6 +2884,7 @@
 src/plugins/fts-squat/Makefile
 src/plugins/lazy-expunge/Makefile
 src/plugins/listescape/Makefile
+src/plugins/mail-filter/Makefile
 src/plugins/mail-log/Makefile
 src/plugins/mailbox-alias/Makefile
 src/plugins/notify/Makefile

--- a/src/plugins/Makefile.am	Fri Nov 22 19:32:06 2013 +0200
+++ b/src/plugins/Makefile.am	Fri Nov 22 19:39:13 2013 +0200
@@ -20,6 +20,7 @@
 	lazy-expunge \
 	listescape \
 	notify \
+	mail-filter \
 	mail-log \
 	mailbox-alias \
 	quota \

--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/src/plugins/mail-filter/Makefile.am	Fri Nov 22 19:39:13 2013 +0200
@@ -0,0 +1,20 @@
+AM_CPPFLAGS = \
+	-I$(top_srcdir)/src/lib \
+	-I$(top_srcdir)/src/lib-mail \
+	-I$(top_srcdir)/src/lib-imap \
+	-I$(top_srcdir)/src/lib-index \
+	-I$(top_srcdir)/src/lib-storage
+
+NOPLUGIN_LDFLAGS =
+lib10_mail_filter_plugin_la_LDFLAGS = -module -avoid-version
+
+module_LTLIBRARIES = \
+	lib10_mail_filter_plugin.la
+
+lib10_mail_filter_plugin_la_SOURCES = \
+	mail-filter-plugin.c \
+	istream-ext-filter.c \
+	ostream-ext-filter.c
+
+noinst_HEADERS = \
+	mail-filter-plugin.h

--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/src/plugins/mail-filter/istream-ext-filter.c	Fri Nov 22 19:39:13 2013 +0200
@@ -0,0 +1,197 @@
+/* Copyright (c) 2013 Dovecot authors, see the included COPYING file */
+
+#include "lib.h"
+#include "str.h"
+#include "net.h"
+#include "eacces-error.h"
+#include "fd-set-nonblock.h"
+#include "ostream.h"
+#include "istream-private.h"
+#include "istream-ext-filter.h"
+
+#include <unistd.h>
+
+struct mail_filter_istream {
+	struct istream_private istream;
+
+	int fd;
+	struct istream *ext_in;
+	struct ostream *ext_out;
+	size_t prev_ret;
+};
+
+static void
+i_stream_mail_filter_close(struct iostream_private *stream, bool close_parent)
+{
+	struct mail_filter_istream *mstream =
+		(struct mail_filter_istream *)stream;
+
+	if (mstream->ext_in != NULL)
+		i_stream_destroy(&mstream->ext_in);
+	if (mstream->ext_out != NULL)
+		o_stream_destroy(&mstream->ext_out);
+	if (mstream->fd != -1) {
+		if (close(mstream->fd) < 0)
+			i_error("ext-filter: close() failed: %m");
+		mstream->fd = -1;
+	}
+	if (close_parent)
+		i_stream_close(mstream->istream.parent);
+}
+
+static ssize_t
+i_stream_read_copy_from(struct istream *istream, struct istream *source)
+{
+	struct istream_private *stream = istream->real_stream;
+	size_t pos;
+	ssize_t ret;
+
+	stream->pos -= stream->skip;
+	stream->skip = 0;
+
+	stream->buffer = i_stream_get_data(source, &pos);
+	if (pos > stream->pos)
+		ret = 0;
+	else do {
+		if ((ret = i_stream_read(source)) == -2)
+			return -2;
+
+		stream->istream.stream_errno = source->stream_errno;
+		stream->istream.eof = source->eof;
+		stream->buffer = i_stream_get_data(source, &pos);
+		/* check again, in case the source stream had been seeked
+		   backwards and the previous read() didn't get us far
+		   enough. */
+	} while (pos <= stream->pos && ret > 0);
+
+	ret = pos > stream->pos ? (ssize_t)(pos - stream->pos) :
+		(ret == 0 ? 0 : -1);
+	stream->pos = pos;
+	i_assert(ret != -1 || stream->istream.eof ||
+		 stream->istream.stream_errno != 0);
+	return ret;
+}
+
+static ssize_t
+i_stream_mail_filter_read_once(struct mail_filter_istream *mstream)
+{
+	struct istream_private *stream = &mstream->istream;
+	ssize_t ret;
+
+	if (mstream->ext_out != NULL) {
+		/* we haven't sent everything yet */
+		(void)o_stream_send_istream(mstream->ext_out, stream->parent);
+		if (mstream->ext_out->stream_errno != 0) {
+			stream->istream.stream_errno =
+				mstream->ext_out->stream_errno;
+			return -1;
+		}
+		if (i_stream_is_eof(stream->parent)) {
+			o_stream_destroy(&mstream->ext_out);
+			/* if we wanted to be a blocking stream,
+			   from now on the rest of the reads are */
+			if (stream->istream.blocking)
+				net_set_nonblock(mstream->fd, FALSE);
+			if (shutdown(mstream->fd, SHUT_WR) < 0)
+				i_error("ext-filter: shutdown() failed: %m");
+		}
+	}
+
+	i_stream_skip(mstream->ext_in, mstream->prev_ret);
+	ret = i_stream_read_copy_from(&stream->istream, mstream->ext_in);
+	mstream->prev_ret = ret < 0 ? 0 : ret;
+	return ret;
+}
+
+static ssize_t i_stream_mail_filter_read(struct istream_private *stream)
+{
+	struct mail_filter_istream *mstream =
+		(struct mail_filter_istream *)stream;
+	ssize_t ret;
+
+	if (mstream->ext_in == NULL) {
+		stream->istream.stream_errno = EIO;
+		return -1;
+	}
+
+	while ((ret = i_stream_mail_filter_read_once(mstream)) == 0) {
+		if (!stream->istream.blocking)
+			break;
+	}
+	return ret;
+}
+
+static int
+i_stream_mail_filter_stat(struct istream_private *stream, bool exact)
+{
+	const struct stat *st;
+
+	i_assert(!exact);
+
+	if (i_stream_stat(stream->parent, exact, &st) < 0)
+		return -1;
+	stream->statbuf = *st;
+	return 0;
+}
+
+static int filter_connect(struct mail_filter_istream *mstream,
+			  const char *socket_path, const char *args)
+{
+	const char **argv;
+	string_t *str;
+	int fd;
+
+	argv = t_strsplit(args, " ");
+
+	if ((fd = net_connect_unix_with_retries(socket_path, 1000)) < 0) {
+		if (errno == EACCES) {
+			i_error("ext-filter: %s",
+				eacces_error_get("net_connect_unix",
+						 socket_path));
+		} else {
+			i_error("ext-filter: net_connect_unix(%s) failed: %m",
+				socket_path);
+		}
+		return -1;
+	}
+	if (mstream->istream.istream.blocking)
+		net_set_nonblock(fd, FALSE);
+
+	mstream->fd = fd;
+	mstream->ext_in =
+		i_stream_create_fd(fd, mstream->istream.max_buffer_size, FALSE);
+	mstream->ext_out = o_stream_create_fd(fd, 0, FALSE);
+
+	str = t_str_new(256);
+	str_append(str, "VERSION\tscript\t3\t0\nnoreply\n");
+	for (; *argv != NULL; argv++) {
+		str_append(str, *argv);
+		str_append_c(str, '\n');
+	}
+	str_append_c(str, '\n');
+
+	o_stream_send(mstream->ext_out, str_data(str), str_len(str));
+	return 0;
+}
+
+struct istream *
+i_stream_create_ext_filter(struct istream *input, const char *socket_path,
+			   const char *args)
+{
+	struct mail_filter_istream *mstream;
+
+	mstream = i_new(struct mail_filter_istream, 1);
+	mstream->istream.iostream.close = i_stream_mail_filter_close;
+	mstream->istream.max_buffer_size = input->real_stream->max_buffer_size;
+	mstream->istream.read = i_stream_mail_filter_read;
+	mstream->istream.stat = i_stream_mail_filter_stat;
+
+	mstream->istream.istream.readable_fd = FALSE;
+	mstream->istream.istream.blocking = input->blocking;
+	mstream->istream.istream.seekable = FALSE;
+
+	mstream->fd = -1;
+	(void)filter_connect(mstream, socket_path, args);
+
+	return i_stream_create(&mstream->istream, input, mstream->fd);
+}

--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/src/plugins/mail-filter/istream-ext-filter.h	Fri Nov 22 19:39:13 2013 +0200
@@ -0,0 +1,8 @@
+#ifndef ISTREAM_MAIL_FILTER_H
+#define ISTREAM_MAIL_FILTER_H
+
+struct istream *
+i_stream_create_ext_filter(struct istream *input, const char *socket_path,
+			   const char *args);
+
+#endif

--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/src/plugins/mail-filter/mail-filter-plugin.c	Fri Nov 22 19:39:13 2013 +0200
@@ -0,0 +1,213 @@
+/* Copyright (c) 2013 Dovecot authors, see the included COPYING file */
+
+#include "lib.h"
+#include "array.h"
+#include "str.h"
+#include "safe-mkstemp.h"
+#include "mail-user.h"
+#include "mail-storage-private.h"
+#include "istream.h"
+#include "istream-seekable.h"
+#include "istream-ext-filter.h"
+#include "ostream-ext-filter.h"
+#include "mail-filter-plugin.h"
+
+/* After buffer grows larger than this, create a temporary file to /tmp
+   where to read the mail. */
+#define MAIL_MAX_MEMORY_BUFFER (1024*128)
+
+#define MAIL_FILTER_MAIL_CONTEXT(obj) \
+	MODULE_CONTEXT(obj, mail_filter_mail_module)
+#define MAIL_FILTER_CONTEXT(obj) \
+	MODULE_CONTEXT(obj, mail_filter_storage_module)
+#define MAIL_FILTER_USER_CONTEXT(obj) \
+	MODULE_CONTEXT(obj, mail_filter_user_module)
+
+struct mail_filter_user {
+	union mail_user_module_context module_ctx;
+
+	const char *socket_path, *args;
+	const char *out_socket_path, *out_args;
+};
+
+const char *mail_filter_plugin_version = DOVECOT_ABI_VERSION;
+
+static MODULE_CONTEXT_DEFINE_INIT(mail_filter_user_module,
+				  &mail_user_module_register);
+static MODULE_CONTEXT_DEFINE_INIT(mail_filter_storage_module,
+				  &mail_storage_module_register);
+static MODULE_CONTEXT_DEFINE_INIT(mail_filter_mail_module,
+				  &mail_module_register);
+
+static int
+mail_filter_mail_save_begin(struct mail_save_context *ctx,
+			    struct istream *input)
+{
+	struct mailbox *box = ctx->transaction->box;
+	struct mail_filter_user *muser =
+		MAIL_FILTER_USER_CONTEXT(box->storage->user);
+	union mailbox_module_context *mbox = MAIL_FILTER_CONTEXT(box);
+	struct ostream *output;
+
+	if (mbox->super.save_begin(ctx, input) < 0)
+		return -1;
+
+	output = o_stream_create_ext_filter(ctx->data.output,
+					    muser->out_socket_path,
+					    muser->out_args);
+	ctx->data.output = output;
+	return 0;
+}
+
+static int seekable_fd_callback(const char **path_r, void *context)
+{
+	struct mail_user *user = context;
+	string_t *path;
+	int fd;
+
+	path = t_str_new(128);
+	mail_user_set_get_temp_prefix(path, user->set);
+	fd = safe_mkstemp(path, 0600, (uid_t)-1, (gid_t)-1);
+	if (fd == -1) {
+		i_error("safe_mkstemp(%s) failed: %m", str_c(path));
+		return -1;
+	}
+
+	/* we just want the fd, unlink it */
+	if (unlink(str_c(path)) < 0) {
+		/* shouldn't happen.. */
+		i_error("unlink(%s) failed: %m", str_c(path));
+		i_close_fd(&fd);
+		return -1;
+	}
+
+	*path_r = str_c(path);
+	return fd;
+}
+
+static int
+mail_filter_istream_opened(struct mail *_mail, struct istream **stream)
+{
+	struct mail_private *mail = (struct mail_private *)_mail;
+	struct mail_user *user = _mail->box->storage->user;
+	struct mail_filter_user *muser = MAIL_FILTER_USER_CONTEXT(user);
+	union mail_module_context *mmail = MAIL_FILTER_MAIL_CONTEXT(mail);
+	struct istream *input, *inputs[2];
+
+	input = *stream;
+	*stream = i_stream_create_ext_filter(input, muser->socket_path,
+					     muser->args);
+	i_stream_unref(&input);
+
+	inputs[0] = *stream;
+	inputs[1] = NULL;
+	*stream = i_stream_create_seekable(inputs, MAIL_MAX_MEMORY_BUFFER,
+					   seekable_fd_callback, user);
+	i_stream_unref(&inputs[0]);
+
+	return mmail->super.istream_opened(_mail, stream);
+}
+
+static void mail_filter_mailbox_allocated(struct mailbox *box)
+{
+	struct mailbox_vfuncs *v = box->vlast;
+	struct mail_filter_user *muser =
+		MAIL_FILTER_USER_CONTEXT(box->storage->user);
+	union mailbox_module_context *mbox;
+	enum mail_storage_class_flags class_flags = box->storage->class_flags;
+
+	mbox = p_new(box->pool, union mailbox_module_context, 1);
+	mbox->super = *v;
+	box->vlast = &mbox->super;
+
+	MODULE_CONTEXT_SET_SELF(box, mail_filter_storage_module, mbox);
+
+	if ((class_flags & MAIL_STORAGE_CLASS_FLAG_OPEN_STREAMS) == 0 &&
+	    (class_flags & MAIL_STORAGE_CLASS_FLAG_BINARY_DATA) != 0 &&
+	    muser->out_socket_path != NULL)
+		v->save_begin = mail_filter_mail_save_begin;
+}
+
+static void mail_filter_mail_allocated(struct mail *_mail)
+{
+	struct mail_private *mail = (struct mail_private *)_mail;
+	struct mail_filter_user *muser =
+		MAIL_FILTER_USER_CONTEXT(_mail->box->storage->user);
+	struct mail_vfuncs *v = mail->vlast;
+	union mail_module_context *mmail;
+
+	mmail = p_new(mail->pool, union mail_module_context, 1);
+	mmail->super = *v;
+	mail->vlast = &mmail->super;
+
+	if (muser->socket_path != NULL)
+		v->istream_opened = mail_filter_istream_opened;
+	MODULE_CONTEXT_SET_SELF(mail, mail_filter_mail_module, mmail);
+}
+
+static void
+mail_filter_parse_setting(struct mail_user *user, const char *name,
+			  const char **socket_path_r, const char **args_r)
+{
+	const char *value, *p;
+
+	value = mail_user_plugin_getenv(user, name);
+	if (value == NULL)
+		return;
+
+	p = strchr(value, ' ');
+	if (p == NULL) {
+		*socket_path_r = p_strdup(user->pool, value);
+		*args_r = "";
+	} else {
+		*socket_path_r = p_strdup_until(user->pool, value, p);
+		*args_r = p_strdup(user->pool, p + 1);
+	}
+	if (**socket_path_r != '/') {
+		/* relative to base_dir */
+		*socket_path_r = p_strdup_printf(user->pool, "%s/%s",
+			user->set->base_dir, *socket_path_r);
+	}
+	if (user->mail_debug) {
+		i_debug("mail_filter: Filtering %s via socket %s",
+			name, *socket_path_r);
+	}
+}
+
+static void mail_filter_mail_user_created(struct mail_user *user)
+{
+	struct mail_user_vfuncs *v = user->vlast;
+	struct mail_filter_user *muser;
+
+	muser = p_new(user->pool, struct mail_filter_user, 1);
+	muser->module_ctx.super = *v;
+	user->vlast = &muser->module_ctx.super;
+
+	mail_filter_parse_setting(user, "mail_filter",
+				  &muser->socket_path, &muser->args);
+	mail_filter_parse_setting(user, "mail_filter_out",
+				  &muser->out_socket_path, &muser->out_args);
+	if (user->mail_debug && muser->socket_path == NULL &&
+	    muser->out_socket_path == NULL) {
+		i_debug("mail_filter and mail_filter_out settings missing, "
+			"ignoring mail_filter plugin");
+	}
+
+	MODULE_CONTEXT_SET(user, mail_filter_user_module, muser);
+}
+
+static struct mail_storage_hooks mail_filter_mail_storage_hooks = {
+	.mail_user_created = mail_filter_mail_user_created,
+	.mailbox_allocated = mail_filter_mailbox_allocated,
+	.mail_allocated = mail_filter_mail_allocated
+};
+
+void mail_filter_plugin_init(struct module *module)
+{
+	mail_storage_hooks_add(module, &mail_filter_mail_storage_hooks);
+}
+
+void mail_filter_plugin_deinit(void)
+{
+	mail_storage_hooks_remove(&mail_filter_mail_storage_hooks);
+}

--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/src/plugins/mail-filter/mail-filter-plugin.h	Fri Nov 22 19:39:13 2013 +0200
@@ -0,0 +1,7 @@
+#ifndef MAIL_FILTER_PLUGIN_H
+#define MAIL_FILTER_PLUGIN_H
+
+void mail_filter_plugin_init(struct module *module);
+void mail_filter_plugin_deinit(void);
+
+#endif

--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/src/plugins/mail-filter/ostream-ext-filter.c	Fri Nov 22 19:39:13 2013 +0200
@@ -0,0 +1,162 @@
+/* Copyright (c) 2013 Dovecot authors, see the included COPYING file */
+
+#include "lib.h"
+#include "str.h"
+#include "net.h"
+#include "eacces-error.h"
+#include "istream.h"
+#include "ostream-private.h"
+#include "ostream-ext-filter.h"
+
+#include <unistd.h>
+
+struct mail_filter_ostream {
+	struct ostream_private ostream;
+
+	int fd;
+	struct istream *ext_in;
+	struct ostream *ext_out;
+	bool flushed;
+};
+
+static void
+o_stream_mail_filter_close(struct iostream_private *stream, bool close_parent)
+{
+	struct mail_filter_ostream *mstream =
+		(struct mail_filter_ostream *)stream;
+
+	if (mstream->ext_in != NULL)
+		i_stream_destroy(&mstream->ext_in);
+	if (mstream->ext_out != NULL)
+		o_stream_destroy(&mstream->ext_out);
+	if (mstream->fd != -1) {
+		if (close(mstream->fd) < 0)
+			i_error("ext-filter: close() failed: %m");
+		mstream->fd = -1;
+	}
+	if (close_parent)
+		o_stream_close(mstream->ostream.parent);
+}
+
+static ssize_t
+o_stream_mail_filter_sendv(struct ostream_private *stream,
+			   const struct const_iovec *iov,
+			   unsigned int iov_count)
+{
+	struct mail_filter_ostream *mstream =
+		(struct mail_filter_ostream *)stream;
+	ssize_t ret;
+
+	if (mstream->ext_out == NULL) {
+		/* connect failed */
+		mstream->ostream.ostream.stream_errno = EIO;
+		return -1;
+	}
+
+	/* send the data to the filter */
+	ret = o_stream_sendv(mstream->ext_out, iov, iov_count);
+	if (ret < 0) {
+		stream->ostream.stream_errno =
+			mstream->ext_out->stream_errno;
+		return -1;
+	}
+	stream->ostream.offset += ret;
+	return ret;
+}
+
+static int o_stream_mail_filter_flush(struct ostream_private *stream)
+{
+	struct mail_filter_ostream *mstream =
+		(struct mail_filter_ostream *)stream;
+	const unsigned char *data;
+	size_t size;
+	ssize_t ret;
+
+	if (mstream->ext_out == NULL) {
+		/* connect failed */
+		return -1;
+	}
+	if (mstream->flushed)
+		return 0;
+
+	if (shutdown(mstream->fd, SHUT_WR) < 0)
+		i_error("ext-filter: shutdown() failed: %m");
+
+	while ((ret = i_stream_read_data(mstream->ext_in, &data, &size, 0)) > 0) {
+		ret = o_stream_send(stream->parent, data, size);
+		if (ret != (ssize_t)size) {
+			i_assert(ret < 0);
+			o_stream_copy_error_from_parent(stream);
+			return -1;
+		}
+		i_stream_skip(mstream->ext_in, size);
+	}
+	i_assert(ret == -1);
+
+	if (mstream->ext_in->stream_errno != 0) {
+		stream->ostream.stream_errno = mstream->ext_in->stream_errno;
+		return -1;
+	}
+
+	ret = o_stream_flush(stream->parent);
+	if (ret < 0)
+		o_stream_copy_error_from_parent(stream);
+	else
+		mstream->flushed = TRUE;
+	return ret;
+}
+
+static int filter_connect(struct mail_filter_ostream *mstream,
+			  const char *socket_path, const char *args)
+{
+	const char **argv;
+	string_t *str;
+	int fd;
+
+	argv = t_strsplit(args, " ");
+
+	if ((fd = net_connect_unix_with_retries(socket_path, 1000)) < 0) {
+		if (errno == EACCES) {
+			i_error("ext-filter: %s",
+				eacces_error_get("net_connect_unix",
+						 socket_path));
+		} else {
+			i_error("ext-filter: net_connect_unix(%s) failed: %m",
+				socket_path);
+		}
+		return -1;
+	}
+	net_set_nonblock(fd, FALSE);
+
+	mstream->fd = fd;
+	mstream->ext_in = i_stream_create_fd(fd, IO_BLOCK_SIZE, FALSE);
+	mstream->ext_out = o_stream_create_fd(fd, 0, FALSE);
+
+	str = t_str_new(256);
+	str_append(str, "VERSION\tscript\t3\t0\nnoreply\n");
+	for (; *argv != NULL; argv++) {
+		str_append(str, *argv);
+		str_append_c(str, '\n');
+	}
+	str_append_c(str, '\n');
+
+	o_stream_send(mstream->ext_out, str_data(str), str_len(str));
+	return 0;
+}
+
+struct ostream *
+o_stream_create_ext_filter(struct ostream *output, const char *socket_path,
+			   const char *args)
+{
+	struct mail_filter_ostream *mstream;
+
+	mstream = i_new(struct mail_filter_ostream, 1);
+	mstream->fd = -1;
+	mstream->ostream.iostream.close = o_stream_mail_filter_close;
+	mstream->ostream.sendv = o_stream_mail_filter_sendv;
+	mstream->ostream.flush = o_stream_mail_filter_flush;
+
+	(void)filter_connect(mstream, socket_path, args);
+
+	return o_stream_create(&mstream->ostream, output, mstream->fd);
+}

--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/src/plugins/mail-filter/ostream-ext-filter.h	Fri Nov 22 19:39:13 2013 +0200
@@ -0,0 +1,8 @@
+#ifndef OSTREAM_MAIL_FILTER_H
+#define OSTREAM_MAIL_FILTER_H
+
+struct ostream *
+o_stream_create_ext_filter(struct ostream *output, const char *socket_path,
+			   const char *args);
+
+#endif