From: Timo Sirainen <timo.sirainen@dovecot.fi>
Date: Sun, 29 Oct 2017 22:44:41 +0000 (+0200)
Subject: login-proxy: Use iostream-proxy API to perform proxying
X-Git-Tag: 2.3.0.rc1~545
X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=5f24bf9ba04644ffe90f9232a7009c18888f5fcc;p=thirdparty%2Fdovecot%2Fcore.git

login-proxy: Use iostream-proxy API to perform proxying
---

diff --git a/src/login-common/login-proxy.c b/src/login-common/login-proxy.c
index f084125001..572f7c3bd6 100644
--- a/src/login-common/login-proxy.c
+++ b/src/login-common/login-proxy.c
@@ -4,11 +4,12 @@
 #include "ioloop.h"
 #include "istream.h"
 #include "ostream.h"
+#include "iostream.h"
+#include "iostream-proxy.h"
 #include "llist.h"
 #include "array.h"
 #include "str.h"
 #include "strescape.h"
-#include "str-sanitize.h"
 #include "time-util.h"
 #include "master-service.h"
 #include "ipc-server.h"
@@ -20,7 +21,7 @@
 
 
 #define MAX_PROXY_INPUT_SIZE 4096
-#define OUTBUF_THRESHOLD 1024
+#define PROXY_MAX_OUTBUF_SIZE 1024
 #define LOGIN_PROXY_DIE_IDLE_SECS 2
 #define LOGIN_PROXY_IPC_PATH "ipc-proxy"
 #define LOGIN_PROXY_IPC_NAME "proxy"
@@ -31,16 +32,20 @@
 #define PROXY_CONNECT_RETRY_MSECS 1000
 #define PROXY_DISCONNECT_INTERVAL_MSECS 100
 
+#define LOGIN_PROXY_SIDE_CLIENT IOSTREAM_PROXY_SIDE_LEFT
+#define LOGIN_PROXY_SIDE_SERVER IOSTREAM_PROXY_SIDE_RIGHT
+
+
 struct login_proxy {
 	struct login_proxy *prev, *next;
 
 	struct client *client;
 	int server_fd;
-	struct io *client_io, *server_io;
+	struct io *client_wait_io, *server_io;
 	struct istream *client_input, *server_input;
 	struct ostream *client_output, *server_output;
+	struct iostream_proxy *iostream_proxy;
 	struct ssl_proxy *ssl_server_proxy;
-	time_t last_io;
 
 	struct timeval created;
 	struct timeout *to, *to_notify;
@@ -82,6 +87,17 @@ static void
 login_proxy_free_delayed(struct login_proxy **_proxy, const char *reason)
 	ATTR_NULL(2);
 
+static time_t proxy_last_io(struct login_proxy *proxy)
+{
+	struct timeval tv1, tv2, tv3, tv4;
+
+	i_stream_get_last_read_time(proxy->client_input, &tv1);
+	i_stream_get_last_read_time(proxy->server_input, &tv2);
+	o_stream_get_last_write_time(proxy->client_output, &tv3);
+	o_stream_get_last_write_time(proxy->server_output, &tv4);
+	return I_MAX(tv1.tv_sec, I_MAX(tv2.tv_sec, I_MAX(tv3.tv_sec, tv4.tv_sec)));
+}
+
 static void login_proxy_free_errstr(struct login_proxy **_proxy,
 				    const char *errstr, bool server)
 {
@@ -93,16 +109,19 @@ static void login_proxy_free_errstr(struct login_proxy **_proxy,
 		str_printfa(reason, ": %s", errstr);
 
 	str_printfa(reason, "(%ds idle, in=%"PRIuUOFF_T", out=%"PRIuUOFF_T,
-		    (int)(ioloop_time - proxy->last_io),
+		    (int)(ioloop_time - proxy_last_io(proxy)),
 		    proxy->server_output->offset, proxy->client_output->offset);
 	if (o_stream_get_buffer_used_size(proxy->client_output) > 0) {
 		str_printfa(reason, "+%"PRIuSIZE_T,
 			    o_stream_get_buffer_used_size(proxy->client_output));
 	}
-	if (proxy->server_io == NULL)
+	if (iostream_proxy_is_waiting_output(proxy->iostream_proxy,
+					     LOGIN_PROXY_SIDE_SERVER))
 		str_append(reason, ", client output blocked");
-	if (proxy->client_io == NULL)
+	if (iostream_proxy_is_waiting_output(proxy->iostream_proxy,
+					     LOGIN_PROXY_SIDE_CLIENT))
 		str_append(reason, ", server output blocked");
+
 	str_append_c(reason, ')');
 	if (server)
 		login_proxy_free_delayed(_proxy, str_c(reason));
@@ -110,81 +129,6 @@ static void login_proxy_free_errstr(struct login_proxy **_proxy,
 		login_proxy_free_reason(_proxy, str_c(reason));
 }
 
-static void login_proxy_free_errno(struct login_proxy **_proxy,
-				   int err, bool server)
-{
-	const char *errstr;
-
-	errstr = err == 0 || err == EPIPE ? "" : strerror(err);
-	login_proxy_free_errstr(_proxy, errstr, server);
-}
-
-static void login_proxy_free_ostream(struct login_proxy **_proxy,
-				     struct ostream *output, bool server)
-{
-	const char *errstr;
-
-	errstr = output->stream_errno == 0 ||
-		output->stream_errno == EPIPE ? "" :
-		o_stream_get_error(output);
-	login_proxy_free_errstr(_proxy, errstr, server);
-}
-
-static void server_input(struct login_proxy *proxy)
-{
-	unsigned char buf[OUTBUF_THRESHOLD];
-	ssize_t ret, ret2;
-
-	proxy->last_io = ioloop_time;
-	if (o_stream_get_buffer_used_size(proxy->client_output) >
-	    OUTBUF_THRESHOLD) {
-		/* client's output buffer is already quite full.
-		   don't send more until we're below threshold. */
-		io_remove(&proxy->server_io);
-		return;
-	}
-
-	ret = net_receive(proxy->server_fd, buf, sizeof(buf));
-	if (ret < 0) {
-		login_proxy_free_errno(&proxy, errno, TRUE);
-		return;
-	}
-	o_stream_cork(proxy->client_output);
-	ret2 = o_stream_send(proxy->client_output, buf, ret);
-	o_stream_uncork(proxy->client_output);
-	if (ret2 != ret)
-		login_proxy_free_ostream(&proxy, proxy->client_output, FALSE);
-}
-
-static void proxy_client_input(struct login_proxy *proxy)
-{
-	const unsigned char *data;
-	size_t size;
-	ssize_t ret;
-
-	proxy->last_io = ioloop_time;
-	if (o_stream_get_buffer_used_size(proxy->server_output) >
-	    OUTBUF_THRESHOLD) {
-		/* proxy's output buffer is already quite full.
-		   don't send more until we're below threshold. */
-		io_remove(&proxy->client_io);
-		return;
-	}
-
-	if (i_stream_read_more(proxy->client_input, &data, &size) < 0) {
-		const char *errstr = i_stream_get_error(proxy->client_input);
-		login_proxy_free_errstr(&proxy, errstr, FALSE);
-		return;
-	}
-	o_stream_cork(proxy->server_output);
-	ret = o_stream_send(proxy->server_output, data, size);
-	o_stream_uncork(proxy->server_output);
-	if (ret != (ssize_t)size)
-		login_proxy_free_ostream(&proxy, proxy->server_output, TRUE);
-	else
-		i_stream_skip(proxy->client_input, ret);
-}
-
 static void proxy_client_disconnected_input(struct login_proxy *proxy)
 {
 	/* we're already disconnected from server. either wait for
@@ -197,44 +141,6 @@ static void proxy_client_disconnected_input(struct login_proxy *proxy)
 	}
 }
 
-static int server_output(struct login_proxy *proxy)
-{
-	proxy->last_io = ioloop_time;
-	if (o_stream_flush(proxy->server_output) < 0) {
-		login_proxy_free_ostream(&proxy, proxy->server_output, TRUE);
-		return 1;
-	}
-
-	if (proxy->client_io == NULL &&
-	    o_stream_get_buffer_used_size(proxy->server_output) <
-	    OUTBUF_THRESHOLD) {
-		/* there's again space in proxy's output buffer, so we can
-		   read more from client. */
-		proxy->client_io = io_add_istream(proxy->client_input,
-						  proxy_client_input, proxy);
-	}
-	return 1;
-}
-
-static int proxy_client_output(struct login_proxy *proxy)
-{
-	proxy->last_io = ioloop_time;
-	if (o_stream_flush(proxy->client_output) < 0) {
-		login_proxy_free_ostream(&proxy, proxy->client_output, FALSE);
-		return 1;
-	}
-
-	if (proxy->server_io == NULL &&
-	    o_stream_get_buffer_used_size(proxy->client_output) <
-	    OUTBUF_THRESHOLD) {
-		/* there's again space in client's output buffer, so we can
-		   read more from proxy. */
-		proxy->server_io =
-			io_add(proxy->server_fd, IO_READ, server_input, proxy);
-	}
-	return 1;
-}
-
 static void proxy_prelogin_input(struct login_proxy *proxy)
 {
 	proxy->callback(proxy->client);
@@ -477,6 +383,8 @@ static void login_proxy_disconnect(struct login_proxy *proxy)
 		proxy->state_rec->num_proxying_connections--;
 	}
 
+	if (proxy->iostream_proxy != NULL)
+		iostream_proxy_unref(&proxy->iostream_proxy);
 	io_remove(&proxy->server_io);
 	i_stream_destroy(&proxy->server_input);
 	o_stream_destroy(&proxy->server_output);
@@ -497,7 +405,7 @@ static void login_proxy_free_final(struct login_proxy *proxy)
 		timeout_remove(&proxy->to);
 	}
 
-	io_remove(&proxy->client_io);
+	io_remove(&proxy->client_wait_io);
 	i_stream_destroy(&proxy->client_input);
 	o_stream_destroy(&proxy->client_output);
 	if (proxy->ssl_server_proxy != NULL) {
@@ -595,11 +503,7 @@ login_proxy_free_full(struct login_proxy **_proxy, const char *reason,
 			ipstr != NULL ? ipstr : "",
 			reason == NULL ? "" : t_strdup_printf(" (%s)", reason),
 			delay_ms == 0 ? "" : t_strdup_printf(" - disconnecting client in %ums", delay_ms)));
-
-		if (proxy->client_io != NULL)
-			io_remove(&proxy->client_io);
 	} else {
-		i_assert(proxy->client_io == NULL);
 		i_assert(proxy->client_input == NULL);
 		i_assert(proxy->client_output == NULL);
 
@@ -611,7 +515,8 @@ login_proxy_free_full(struct login_proxy **_proxy, const char *reason,
 	if (delay_ms == 0)
 		login_proxy_free_final(proxy);
 	else {
-		proxy->client_io = io_add_istream(proxy->client_input,
+		i_assert(proxy->client_wait_io == NULL);
+		proxy->client_wait_io = io_add_istream(proxy->client_input,
 			proxy_client_disconnected_input, proxy);
 	}
 
@@ -678,6 +583,37 @@ login_proxy_get_ssl_flags(const struct login_proxy *proxy)
 	return proxy->ssl_flags;
 }
 
+static void
+login_proxy_finished(enum iostream_proxy_side side,
+		     enum iostream_proxy_status status,
+		     struct login_proxy *proxy)
+{
+	const char *errstr;
+	bool server_side;
+
+	server_side = side == LOGIN_PROXY_SIDE_SERVER;
+	switch (status) {
+	case IOSTREAM_PROXY_STATUS_INPUT_EOF:
+		/* success */
+		errstr = "";
+		break;
+	case IOSTREAM_PROXY_STATUS_INPUT_ERROR:
+		errstr = side == LOGIN_PROXY_SIDE_CLIENT ?
+			i_stream_get_error(proxy->client_input) :
+			i_stream_get_error(proxy->server_input);
+		break;
+	case IOSTREAM_PROXY_STATUS_OTHER_SIDE_OUTPUT_ERROR:
+		server_side = !server_side;
+		errstr = side == LOGIN_PROXY_SIDE_CLIENT ?
+			o_stream_get_error(proxy->server_output) :
+			o_stream_get_error(proxy->client_output);
+		break;
+	default:
+		i_unreached();
+	}
+	login_proxy_free_errstr(&proxy, errstr, server_side);
+}
+
 static void login_proxy_notify(struct login_proxy *proxy)
 {
 	login_proxy_state_notify(proxy_state, proxy->client->proxy_user);
@@ -686,8 +622,6 @@ static void login_proxy_notify(struct login_proxy *proxy)
 void login_proxy_detach(struct login_proxy *proxy)
 {
 	struct client *client = proxy->client;
-	const unsigned char *data;
-	size_t size;
 
 	pool_unref(&proxy->client->preproxy_pool);
 
@@ -696,30 +630,25 @@ void login_proxy_detach(struct login_proxy *proxy)
 	i_assert(proxy->server_output != NULL);
 
 	timeout_remove(&proxy->to);
+	io_remove(&proxy->server_io);
 
 	proxy->detached = TRUE;
 	proxy->client_input = client->input;
 	proxy->client_output = client->output;
 
+	i_stream_set_persistent_buffers(proxy->server_input, FALSE);
 	i_stream_set_persistent_buffers(client->input, FALSE);
-	o_stream_set_max_buffer_size(client->output, (size_t)-1);
-	o_stream_set_flush_callback(client->output, proxy_client_output, proxy);
+	o_stream_set_max_buffer_size(client->output, PROXY_MAX_OUTBUF_SIZE);
 	client->input = NULL;
 	client->output = NULL;
 
-	/* send all pending client input to proxy */
-	data = i_stream_get_data(proxy->client_input, &size);
-	if (size != 0)
-		o_stream_nsend(proxy->server_output, data, size);
-
 	/* from now on, just do dummy proxying */
-	io_remove(&proxy->server_io);
-	proxy->server_io =
-		io_add(proxy->server_fd, IO_READ, server_input, proxy);
-	proxy->client_io =
-		io_add_istream(proxy->client_input, proxy_client_input, proxy);
-	o_stream_set_flush_callback(proxy->server_output, server_output, proxy);
-	i_stream_destroy(&proxy->server_input);
+	proxy->iostream_proxy =
+		iostream_proxy_create(proxy->client_input, proxy->client_output,
+				      proxy->server_input, proxy->server_output);
+	iostream_proxy_set_completion_callback(proxy->iostream_proxy,
+					       login_proxy_finished, proxy);
+	iostream_proxy_start(proxy->iostream_proxy);
 
 	if (proxy->notify_refresh_secs != 0) {
 		proxy->to_notify =
@@ -811,12 +740,13 @@ void login_proxy_kill_idle(void)
 
 	for (proxy = login_proxies; proxy != NULL; proxy = next) {
 		next = proxy->next;
+		time_t last_io = proxy_last_io(proxy);
 
-		if (proxy->last_io <= stop_timestamp)
+		if (last_io <= stop_timestamp)
 			proxy_kill_idle(proxy);
 		else {
 			i_assert(proxy->to == NULL);
-			stop_msecs = (proxy->last_io - stop_timestamp) * 1000;
+			stop_msecs = (last_io - stop_timestamp) * 1000;
 			proxy->to = timeout_add(stop_msecs,
 						proxy_kill_idle, proxy);
 		}