lmtp client, proxy: Several bugfixes.

--HG--
branch : HEAD

diff --git a/src/lib-lda/lmtp-client.c b/src/lib-lda/lmtp-client.c
index 5b2dd1a16bed4337a1da5831ab1dddccd15bcc6b..4eaaf01d90c3e6a391ccdd91e2626090df2e1f78 100644 (file)
--- a/src/lib-lda/lmtp-client.c
+++ b/src/lib-lda/lmtp-client.c
@@ -53,6 +53,8 @@ struct lmtp_client {
        unsigned int rcpt_next_send_idx;
        struct istream *data_input;
        unsigned char output_last;
+
+       unsigned int output_finished:1;
 };
 
 static void lmtp_client_send_rcpts(struct lmtp_client *client);
@@ -124,36 +126,52 @@ static void lmtp_client_fail(struct lmtp_client *client, const char *line)
 static bool
 lmtp_client_rcpt_next(struct lmtp_client *client, const char *line)
 {
-       struct lmtp_rcpt *recipients;
-       unsigned int i, count;
+       struct lmtp_rcpt *rcpt;
        bool success, all_sent;
 
        success = line[0] == '2';
 
-       recipients = array_get_modifiable(&client->recipients, &count);
-       for (i = client->rcpt_next_receive_idx; i < count; i++) {
-               recipients[i].failed = !success;
-               recipients[i].rcpt_to_callback(success, line,
-                                              recipients[i].context);
-       }
-       all_sent = i == client->rcpt_next_receive_idx;
-       client->rcpt_next_receive_idx = i;
+       rcpt = array_idx_modifiable(&client->recipients,
+                                   client->rcpt_next_receive_idx);
+       rcpt->failed = !success;
+       rcpt->rcpt_to_callback(success, line, rcpt->context);
+
+       all_sent = ++client->rcpt_next_receive_idx ==
+               array_count(&client->recipients);
        return all_sent && client->data_input != NULL;
 }
 
-static int
+static bool
 lmtp_client_data_next(struct lmtp_client *client, const char *line)
 {
        struct lmtp_rcpt *rcpt;
+       unsigned int i, count;
        bool last;
 
-       rcpt = array_idx_modifiable(&client->recipients,
-                                   client->rcpt_next_data_idx);
-       rcpt->failed = line[0] != '2';
-       last = ++client->rcpt_next_data_idx == array_count(&client->recipients);
+       switch (client->protocol) {
+       case LMTP_CLIENT_PROTOCOL_SMTP:
+               i_assert(client->rcpt_next_data_idx == 0);
 
-       rcpt->data_callback(!rcpt->failed, line, rcpt->context);
-       return last ? -1 : 0;
+               rcpt = array_get_modifiable(&client->recipients, &count);
+               for (i = 0; i < count; i++) {
+                       rcpt[i].failed = line[0] != '2';
+                       rcpt[i].data_callback(!rcpt->failed, line,
+                                             rcpt[i].context);
+               }
+               client->rcpt_next_data_idx = count;
+               last = TRUE;
+               break;
+       case LMTP_CLIENT_PROTOCOL_LMTP:
+               rcpt = array_idx_modifiable(&client->recipients,
+                                           client->rcpt_next_data_idx);
+               rcpt->failed = line[0] != '2';
+               last = ++client->rcpt_next_data_idx ==
+                       array_count(&client->recipients);
+
+               rcpt->data_callback(!rcpt->failed, line, rcpt->context);
+               break;
+       }
+       return !last;
 }
 
 static void lmtp_client_send_data(struct lmtp_client *client)
@@ -163,6 +181,9 @@ static void lmtp_client_send_data(struct lmtp_client *client)
        size_t i, size;
        int ret;
 
+       if (client->output_finished)
+               return;
+
        while ((ret = i_stream_read_data(client->data_input,
                                         &data, &size, 0)) > 0) {
                add = '\0';
@@ -216,6 +237,7 @@ static void lmtp_client_send_data(struct lmtp_client *client)
                (void)o_stream_send(client->output, "\r\n", 2);
        }
        (void)o_stream_send(client->output, ".\r\n", 3);
+       client->output_finished = TRUE;
 }
 
 static void lmtp_client_send_handshake(struct lmtp_client *client)
@@ -289,8 +311,9 @@ static int lmtp_client_input_line(struct lmtp_client *client, const char *line)
        case LMTP_INPUT_STATE_RCPT_TO:
                if (!lmtp_client_rcpt_next(client, line))
                        break;
-               /* fall through */
                client->input_state++;
+               o_stream_send_str(client->output, "DATA\r\n");
+               break;
        case LMTP_INPUT_STATE_DATA_CONTINUE:
                /* Start sending DATA */
                if (strncmp(line, "354", 3) != 0) {
@@ -304,7 +327,7 @@ static int lmtp_client_input_line(struct lmtp_client *client, const char *line)
                break;
        case LMTP_INPUT_STATE_DATA:
                /* DATA replies */
-               if (lmtp_client_data_next(client, line) < 0)
+               if (!lmtp_client_data_next(client, line))
                        return -1;
                break;
        }
@@ -334,7 +357,8 @@ static void lmtp_client_wait_connect(struct lmtp_client *client)
        if (err != 0) {
                i_error("lmtp client: connect(%s, %u) failed: %s",
                        client->host, client->port, strerror(err));
-               lmtp_client_fail(client, NULL);
+               lmtp_client_fail(client, ERRSTR_TEMP_REMOTE_FAILURE
+                                " (connect)");
                return;
        }
        io_remove(&client->io);
@@ -348,7 +372,8 @@ static int lmtp_client_output(struct lmtp_client *client)
 
        o_stream_cork(client->output);
        if ((ret = o_stream_flush(client->output)) < 0)
-               lmtp_client_fail(client, NULL);
+               lmtp_client_fail(client, ERRSTR_TEMP_REMOTE_FAILURE
+                                " (disconnected in output)");
        else if (client->input_state == LMTP_INPUT_STATE_DATA)
                lmtp_client_send_data(client);
        o_stream_uncork(client->output);
@@ -415,8 +440,13 @@ void lmtp_client_add_rcpt(struct lmtp_client *client, const char *address,
 
 void lmtp_client_send(struct lmtp_client *client, struct istream *data_input)
 {
+       i_stream_ref(data_input);
        client->data_input = data_input;
-       o_stream_send_str(client->output, "DATA\r\n");
+
+       if (client->rcpt_next_receive_idx == array_count(&client->recipients)) {
+               client->input_state++;
+               o_stream_send_str(client->output, "DATA\r\n");
+       }
 }
 
 void lmtp_client_send_more(struct lmtp_client *client)

diff --git a/src/lib-lda/lmtp-client.h b/src/lib-lda/lmtp-client.h
index 76623fdb6a9fdd0e0e298fd72025aa0fceff0e18..ac9ce78927d94710bfa40757c0411133c299244c 100644 (file)
--- a/src/lib-lda/lmtp-client.h
+++ b/src/lib-lda/lmtp-client.h
@@ -1,6 +1,8 @@
 #ifndef LMTP_CLIENT_H
 #define LMTP_CLIENT_H
 
+#define ERRSTR_TEMP_REMOTE_FAILURE "451 4.4.0 Remote server not answering"
+
 /* LMTP/SMTP client code. */
 
 enum lmtp_client_protocol {

diff --git a/src/lmtp/commands.c b/src/lmtp/commands.c
index f193afe9b6f5ee12069369f8b2836ba39fa9585f..9b961e52d23846500c1367f7489a16603005f870 100644 (file)
--- a/src/lmtp/commands.c
+++ b/src/lmtp/commands.c
@@ -24,6 +24,8 @@
 #define ERRSTR_TEMP_MAILBOX_FAIL "451 4.3.0 <%s> Temporary internal error"
 #define ERRSTR_TEMP_USERDB_FAIL "451 4.3.0 <%s> Temporary user lookup failure"
 
+#define LMTP_PROXY_DEFAULT_TIMEOUT_MSECS (1000*30)
+
 int cmd_lhlo(struct client *client, const char *args ATTR_UNUSED)
 {
        client_state_reset(client);
@@ -94,7 +96,7 @@ client_proxy_rcpt_parse_fields(struct lmtp_proxy_settings *set,
                               const char *const *args, const char **address)
 {
        const char *p, *key, *value;
-       bool proxying = FALSE;
+       bool proxying = FALSE, port_set = FALSE;
 
        for (; *args != NULL; args++) {
                p = strchr(*args, '=');
@@ -110,9 +112,23 @@ client_proxy_rcpt_parse_fields(struct lmtp_proxy_settings *set,
                        proxying = TRUE;
                else if (strcmp(key, "host") == 0)
                        set->host = value;
-               else if (strcmp(key, "port") == 0)
+               else if (strcmp(key, "port") == 0) {
                        set->port = atoi(value);
-               else if (strcmp(key, "user") == 0) {
+                       port_set = TRUE;
+               } else if (strcmp(key, "proxy_timeout") == 0)
+                       set->timeout_msecs = atoi(value)*1000;
+               else if (strcmp(key, "protocol") == 0) {
+                       if (strcmp(value, "lmtp") == 0)
+                               set->protocol = LMTP_CLIENT_PROTOCOL_LMTP;
+                       else if (strcmp(value, "smtp") == 0) {
+                               set->protocol = LMTP_CLIENT_PROTOCOL_SMTP;
+                               if (!port_set)
+                                       set->port = 25;
+                       } else {
+                               i_error("proxy: Unknown protocol %s", value);
+                               return FALSE;
+                       }
+               } else if (strcmp(key, "user") == 0) {
                        /* changing the username */
                        *address = value;
                } else {
@@ -176,6 +192,9 @@ static bool client_proxy_rcpt(struct client *client, const char *address)
 
        memset(&set, 0, sizeof(set));
        set.port = client->local_port;
+       set.protocol = LMTP_CLIENT_PROTOCOL_LMTP;
+       set.timeout_msecs = LMTP_PROXY_DEFAULT_TIMEOUT_MSECS;
+
        if (!client_proxy_rcpt_parse_fields(&set, fields, &address)) {
                /* not proxying this user */
                pool_unref(&pool);

diff --git a/src/lmtp/lmtp-proxy.c b/src/lmtp/lmtp-proxy.c
index 8af3dba88a831a6613b4452dc99feffb7f0b503e..279ebfcc20e223900613bfb1083caa11a236a800 100644 (file)
--- a/src/lmtp/lmtp-proxy.c
+++ b/src/lmtp/lmtp-proxy.c
@@ -10,7 +10,6 @@
 #include "lmtp-proxy.h"
 
 #define LMTP_MAX_LINE_LEN 1024
-#define LMTP_PROXY_OUTPUT_TIMEOUT_MSECS 1000
 
 struct lmtp_proxy_recipient {
        struct lmtp_proxy_connection *conn;
@@ -47,13 +46,16 @@ struct lmtp_proxy {
        struct ostream *client_output;
        struct tee_istream *tee_data_input;
 
+       unsigned int max_timeout_msecs;
+
        void (*finish_callback)(void *);
        void *finish_context;
 
        unsigned int finished:1;
 };
 
-static void lmtp_proxy_conn_deinit(struct lmtp_proxy_connection *conn);
+static void lmtp_proxy_conn_deinit(struct lmtp_proxy_connection *conn,
+                                  const char *reason);
 static void lmtp_proxy_data_input(struct lmtp_proxy *proxy);
 
 struct lmtp_proxy *
@@ -81,7 +83,7 @@ static void lmtp_proxy_connections_deinit(struct lmtp_proxy *proxy)
 
        conns = array_get(&proxy->connections, &count);
        for (i = 0; i < count; i++)
-               lmtp_proxy_conn_deinit(conns[i]);
+               lmtp_proxy_conn_deinit(conns[i], "451 4.3.0 Aborting");
 }
 
 void lmtp_proxy_deinit(struct lmtp_proxy **_proxy)
@@ -116,6 +118,8 @@ lmtp_proxy_get_connection(struct lmtp_proxy *proxy,
        struct lmtp_proxy_connection *const *conns, *conn;
        unsigned int i, count;
 
+       i_assert(set->timeout_msecs > 0);
+
        conns = array_get(&proxy->connections, &count);
        for (i = 0; i < count; i++) {
                if (conns[i]->set.port == set->port &&
@@ -130,13 +134,17 @@ lmtp_proxy_get_connection(struct lmtp_proxy *proxy,
        conn->set.timeout_msecs = set->timeout_msecs;
        array_append(&proxy->connections, &conn, 1);
        conn->client = lmtp_client_init(proxy->mail_from, proxy->my_hostname);
-       if (lmtp_client_connect_tcp(conn->client, LMTP_CLIENT_PROTOCOL_LMTP,
+       if (lmtp_client_connect_tcp(conn->client, set->protocol,
                                    conn->set.host, conn->set.port) < 0)
                conn->failed = TRUE;
+
+       if (proxy->max_timeout_msecs < set->timeout_msecs)
+               proxy->max_timeout_msecs = set->timeout_msecs;
        return conn;
 }
 
-static void lmtp_proxy_conn_deinit(struct lmtp_proxy_connection *conn)
+static void lmtp_proxy_conn_deinit(struct lmtp_proxy_connection *conn,
+                                  const char *reason)
 {
        struct lmtp_proxy_recipient *rcpt;
        unsigned int i, count;
@@ -145,7 +153,7 @@ static void lmtp_proxy_conn_deinit(struct lmtp_proxy_connection *conn)
        rcpt = array_get_modifiable(&conn->proxy->rcpt_to, &count);
        for (i = 0; i < count; i++) {
                if (rcpt[i].conn == conn && !rcpt[i].rcpt_to_failed)
-                       rcpt[i].reply = ERRSTR_TEMP_REMOTE_FAILURE;
+                       rcpt[i].reply = reason;
        }
 
        if (conn->client != NULL)
@@ -175,34 +183,37 @@ static bool lmtp_proxy_send_replies(struct lmtp_proxy *proxy)
 }
 
 static void lmtp_proxy_finish(struct lmtp_proxy *proxy)
+{
+       i_assert(!proxy->finished);
+
+       proxy->finished = TRUE;
+       proxy->finish_callback(proxy->finish_context);
+}
+
+static void lmtp_proxy_try_finish(struct lmtp_proxy *proxy)
+{
+       if (lmtp_proxy_send_replies(proxy))
+               lmtp_proxy_finish(proxy);
+}
+
+static void lmtp_proxy_data_disconnected(struct lmtp_proxy *proxy)
 {
        struct lmtp_proxy_recipient *rcpt;
        unsigned int i, count;
        bool ret;
 
-       i_assert(!proxy->finished);
-
-       /* if we haven't sent something yet, they're failures */
        rcpt = array_get_modifiable(&proxy->rcpt_to, &count);
        for (i = proxy->rcpt_next_reply_idx; i < count; i++) {
                if (!rcpt[i].rcpt_to_failed) {
                        i_assert(!rcpt[i].data_reply_received);
-                       rcpt[i].reply = ERRSTR_TEMP_REMOTE_FAILURE;
+                       rcpt[i].reply = "451 4.4.0 Client disconnected in DATA";
                        rcpt[i].data_reply_received = TRUE;
                }
        }
-
        ret = lmtp_proxy_send_replies(proxy);
        i_assert(ret);
 
-       proxy->finished = TRUE;
-       proxy->finish_callback(proxy->finish_context);
-}
-
-static void lmtp_proxy_try_finish(struct lmtp_proxy *proxy)
-{
-       if (lmtp_proxy_send_replies(proxy))
-               lmtp_proxy_finish(proxy);
+       lmtp_proxy_finish(proxy);
 }
 
 static void
@@ -238,9 +249,6 @@ lmtp_proxy_conn_data(bool success ATTR_UNUSED, const char *reply, void *context)
 
        i_assert(conn->proxy->data_input != NULL);
 
-       if (reply == NULL)
-               reply = ERRSTR_TEMP_REMOTE_FAILURE;
-
        rcpt = array_get_modifiable(&conn->proxy->rcpt_to, &count);
        for (i = conn->data_next_reply_low_idx; i < count; i++) {
                if (rcpt[i].conn == conn && !rcpt[i].rcpt_to_failed) {
@@ -293,7 +301,8 @@ static void lmtp_proxy_output_timeout(struct lmtp_proxy *proxy)
        }
        i_assert(max_conn != NULL);
 
-       lmtp_proxy_conn_deinit(max_conn);
+       lmtp_proxy_conn_deinit(max_conn, ERRSTR_TEMP_REMOTE_FAILURE
+                              " (timeout)");
 }
 
 static void lmtp_proxy_wait_for_output(struct lmtp_proxy *proxy)
@@ -303,12 +312,12 @@ static void lmtp_proxy_wait_for_output(struct lmtp_proxy *proxy)
        if (proxy->io != NULL)
                io_remove(&proxy->io);
        if (array_count(&proxy->connections) > 1) {
-               proxy->to = timeout_add(LMTP_PROXY_OUTPUT_TIMEOUT_MSECS,
+               proxy->to = timeout_add(proxy->max_timeout_msecs,
                                        lmtp_proxy_output_timeout, proxy);
        }
 }
 
-static bool lmtp_proxy_read_data(struct lmtp_proxy *proxy)
+static bool lmtp_proxy_data_read(struct lmtp_proxy *proxy)
 {
        size_t size;
 
@@ -318,8 +327,12 @@ static bool lmtp_proxy_read_data(struct lmtp_proxy *proxy)
                lmtp_proxy_wait_for_output(proxy);
                return FALSE;
        case -1:
-               /* disconnected */
-               lmtp_proxy_finish(proxy);
+               if (proxy->data_input->stream_errno != 0)
+                       lmtp_proxy_data_disconnected(proxy);
+               else {
+                       /* finished reading data input. now we'll just have to
+                          wait for replies. */
+               }
                return FALSE;
        case 0:
                /* nothing new read */
@@ -346,7 +359,7 @@ static void lmtp_proxy_data_input(struct lmtp_proxy *proxy)
                conns = array_get(&proxy->connections, &count);
                for (i = 0; i < count; i++)
                        lmtp_client_send_more(conns[i]->client);
-       } while (lmtp_proxy_read_data(proxy));
+       } while (lmtp_proxy_data_read(proxy));
 }
 
 void lmtp_proxy_start(struct lmtp_proxy *proxy, struct istream *data_input,
@@ -364,6 +377,7 @@ void lmtp_proxy_start(struct lmtp_proxy *proxy, struct istream *data_input,
        for (i = 0; i < count; i++) {
                conns[i]->data_input =
                        tee_i_stream_create_child(proxy->tee_data_input);
+               lmtp_client_send(conns[i]->client, conns[i]->data_input);
        }
 
        lmtp_proxy_data_input(proxy);

diff --git a/src/lmtp/lmtp-proxy.h b/src/lmtp/lmtp-proxy.h
index 200827bca7714592915c62ecd7e366c017391a82..715811ce158fd5868c748743bab1e4e438983bdf 100644 (file)
--- a/src/lmtp/lmtp-proxy.h
+++ b/src/lmtp/lmtp-proxy.h
@@ -2,13 +2,13 @@
 #define LMTP_PROXY_H
 
 #include "network.h"
-
-#define ERRSTR_TEMP_REMOTE_FAILURE "451 4.4.0 Remote server not answering"
+#include "lmtp-client.h"
 
 struct lmtp_proxy_settings {
        const char *host;
        unsigned int port;
        unsigned int timeout_msecs;
+       enum lmtp_client_protocol protocol;
 };
 
 struct lmtp_proxy *