Fixes skupperproject#1446: check for raw read close only after draining read buffers (skupperproject#1447)

When polling the raw connection it is possible for
pn_raw_connection_is_read_closed() to return true when there are still
read buffers available. This can happen when the read buffers are not
completely drained due to flow control. The patch moves the check for
raw read closed to after the point where all read buffers have been
drained.

Closes skupperproject#1446

Author: kgiusti

src/adaptors/adaptor_tls.c

@@ -58,9 +58,10 @@ struct qd_tls_t {
     qd_tls_on_secure_cb_t *on_secure_cb;
     bool                   tls_has_output;
     bool                   tls_error;
-    bool                   output_eos;      // pn_tls_close_output() called
-    bool                   input_drained;   // no more decrypted output, raw conn read closed
-    bool                   output_flushed;  // encrypt done, raw conn write closed
+    bool                   output_eos;       // pn_tls_close_output() called
+    bool                   raw_read_drained; // raw conn read closed and all buffer read
+    bool                   input_drained;    // no more decrypted output, raw conn read closed
+    bool                   output_flushed;   // encrypt done, raw conn write closed
 
     uint64_t encrypted_output_bytes;
     uint64_t encrypted_input_bytes;
@@ -1274,8 +1275,13 @@ int qd_tls_do_io2(qd_tls_t                      *tls,
         if (capacity > 0 && (!pn_tls_is_secure(tls->tls_session) || input_data != 0)) {
             size_t pushed = 0;
             total_octets  = 0;
-            while (pushed < capacity && pn_raw_connection_take_read_buffers(raw_conn, &pn_buf_desc, 1) == 1) {
-                if (pn_buf_desc.size) {
+            while (pushed < capacity) {
+                size_t took = pn_raw_connection_take_read_buffers(raw_conn, &pn_buf_desc, 1);
+                if (took != 1) {
+                    // No more read buffers available. Now it is safe to check if the raw connection has closed
+                    tls->raw_read_drained = pn_raw_connection_is_read_closed(raw_conn);
+                    break;
+                } else if (pn_buf_desc.size) {
                     total_octets += pn_buf_desc.size;
                     given = pn_tls_give_decrypt_input_buffers(tls->tls_session, &pn_buf_desc, 1);
                     assert(given == 1);
@@ -1427,6 +1433,8 @@ int qd_tls_do_io2(qd_tls_t                      *tls,
         return -1;
     }
 
+    // check if the output (encrypt) side is done
+    //
     if (tls->output_eos && !pn_tls_is_encrypt_output_pending(tls->tls_session)) {
         // We closed the encrypt side of the TLS connection and we've sent all output
         tls->output_flushed = true;
@@ -1437,9 +1445,12 @@ int qd_tls_do_io2(qd_tls_t                      *tls,
         }
     }
 
-    if (pn_tls_is_input_closed(tls->tls_session)
-        || (pn_raw_connection_is_read_closed(raw_conn) && !pn_tls_is_decrypt_output_pending(tls->tls_session))) {
-        // TLS will not take any more encrypted input - drain the raw conn input
+    // check for end of input (decrypt done)
+    //
+    if (pn_tls_is_input_closed(tls->tls_session)) {
+        // TLS clean close signalled by remote. Do not read any more data (prevent truncation attack).
+        //
+        tls->input_drained = true;
         if (!pn_raw_connection_is_read_closed(raw_conn)) {
             qd_log(tls->log_module, QD_LOG_DEBUG,
                    "[C%" PRIu64 "] TLS input closed - closing read side of raw connection", tls->conn_id);
@@ -1448,8 +1459,12 @@ int qd_tls_do_io2(qd_tls_t                      *tls,
         while (pn_raw_connection_take_read_buffers(raw_conn, &pn_buf_desc, 1) == 1) {
             qd_buffer_free(_pn_buf_desc_take_buffer(&pn_buf_desc));
         }
-
+    } else if (tls->raw_read_drained && !pn_tls_is_decrypt_output_pending(tls->tls_session)) {
+        // Unclean close: remote simply dropped the connection. Done reading remaining decrypted output.
+        //
         tls->input_drained = true;
+        qd_log(tls->log_module, QD_LOG_DEBUG,
+               "[C%" PRIu64 "] TLS input closed", tls->conn_id);
     }
 
     return tls->input_drained && tls->output_flushed ? QD_TLS_DONE : 0;

src/adaptors/tcp_lite/tcp_lite.c

@@ -713,13 +713,13 @@ static void grant_read_buffers_XSIDE_IO(tcplite_connection_t *conn, const size_t
 }
 
 
-static uint64_t produce_read_buffers_XSIDE_IO(tcplite_connection_t *conn, qd_message_t *stream, bool *blocked)
+static uint64_t produce_read_buffers_XSIDE_IO(tcplite_connection_t *conn, qd_message_t *stream, bool *read_closed)
 {
     ASSERT_RAW_IO;
     uint64_t octet_count = 0;
+    *read_closed = false;
 
     if (qd_message_can_produce_buffers(stream)) {
-        *blocked = false;
         qd_buffer_list_t qd_buffers = DEQ_EMPTY;
         pn_raw_buffer_t  raw_buffers[RAW_BUFFER_BATCH_SIZE];
         size_t           count;
@@ -742,12 +742,15 @@ static uint64_t produce_read_buffers_XSIDE_IO(tcplite_connection_t *conn, qd_mes
             count = pn_raw_connection_take_read_buffers(conn->raw_conn, raw_buffers, RAW_BUFFER_BATCH_SIZE);
         }
 
+        // ISSUE-1446: it is only safe to check pn_raw_connection_is_read_closed() after all read buffers are drained since
+        // the connection can be marked closed while read buffers are still pending in the raw connection.
+        //
+        *read_closed = pn_raw_connection_is_read_closed(conn->raw_conn);
+
         if (!DEQ_IS_EMPTY(qd_buffers)) {
             //qd_log(LOG_TCP_ADAPTOR, QD_LOG_DEBUG, "[C%"PRIu64"] produce_read_buffers_XSIDE_IO - Producing %ld buffers", conn->conn_id, DEQ_SIZE(qd_buffers));
             qd_message_produce_buffers(stream, &qd_buffers);
         }
-    } else {
-        *blocked = true;
     }
 
     return octet_count;
@@ -1240,14 +1243,14 @@ static bool manage_flow_XSIDE_IO(tcplite_connection_t *conn)
         //
         // Produce available read buffers into the inbound stream
         //
+        bool read_closed = false;
         bool was_blocked = window_full(conn);
-        bool blocked;
-        uint64_t octet_count = produce_read_buffers_XSIDE_IO(conn, conn->inbound_stream, &blocked);
+        uint64_t octet_count = produce_read_buffers_XSIDE_IO(conn, conn->inbound_stream, &read_closed);
         conn->inbound_octets += octet_count;
 
         if (octet_count > 0) {
             qd_log(LOG_TCP_ADAPTOR, QD_LOG_DEBUG, "[C%"PRIu64"] %cSIDE Raw read: Produced %"PRIu64" octets into stream", conn->conn_id, conn->listener_side ? 'L' : 'C', octet_count);
-            if (!was_blocked && window_full(conn)) {
+            if (!was_blocked && window_full(conn) && !read_closed) {
                 conn->window.closed_count += 1;
                 qd_log(LOG_TCP_ADAPTOR, QD_LOG_DEBUG, DLV_FMT " TCP RX window CLOSED: inbound_bytes=%" PRIu64 " unacked=%" PRIu64,
                        DLV_ARGS(conn->inbound_delivery), conn->inbound_octets,
@@ -1271,8 +1274,8 @@ static bool manage_flow_XSIDE_IO(tcplite_connection_t *conn)
         // If the raw connection is read-closed and the last produce did not block, settle and complete
         // the inbound stream/delivery and close out the inbound half of the connection.
         //
-        if (pn_raw_connection_is_read_closed(conn->raw_conn) && !blocked) {
-            qd_log(LOG_TCP_ADAPTOR, QD_LOG_DEBUG, "[C%"PRIu64"] Read-closed - close inbound delivery", conn->conn_id);
+        if (read_closed) {
+            qd_log(LOG_TCP_ADAPTOR, QD_LOG_DEBUG, DLV_FMT " Raw conn read-closed - close inbound delivery", DLV_ARGS(conn->inbound_delivery));
             qd_message_set_receive_complete(conn->inbound_stream);
             qdr_delivery_continue(tcplite_context->core, conn->inbound_delivery, false);
             qdr_delivery_set_context(conn->inbound_delivery, 0);