tls: fix regression and proper shutdown

include/fluent-bit/tls/flb_tls.h

@@ -84,6 +84,7 @@ struct flb_tls_backend {
 
     /* Session management */
     void *(*session_create) (struct flb_tls *, int);
+    void (*session_invalidate) (void *);
     int (*session_destroy) (void *);
     const char *(*session_alpn_get) (void *);
 
@@ -144,6 +145,7 @@ int flb_tls_set_ciphers(struct flb_tls *tls, const char *ciphers);
 struct mk_list *flb_tls_get_config_map(struct flb_config *config);
 
 int flb_tls_session_destroy(struct flb_tls_session *session);
+int flb_tls_session_invalidate(struct flb_tls_session *session);
 
 int flb_tls_session_create(struct flb_tls *tls,
                            struct flb_connection *connection,

src/flb_upstream.c

@@ -515,6 +515,11 @@ static int prepare_destroy_conn(struct flb_connection *u_conn)
     }
 
     if (u_conn->fd > 0) {
+#ifdef FLB_HAVE_TLS
+        if (u_conn->tls_session != NULL) {
+            flb_tls_session_invalidate(u_conn->tls_session);
+        }
+#endif
         shutdown_connection(u_conn);
 
         flb_socket_close(u_conn->fd);

src/tls/flb_tls.c

@@ -766,3 +766,16 @@ int flb_tls_session_destroy(struct flb_tls_session *session)
 
     return 0;
 }
+
+int flb_tls_session_invalidate(struct flb_tls_session *session)
+{
+    if (session == NULL || session->tls == NULL) {
+        return -1;
+    }
+
+    if (session->ptr != NULL && session->tls->api->session_invalidate != NULL) {
+        session->tls->api->session_invalidate(session->ptr);
+    }
+
+    return 0;
+}

src/tls/openssl.c

@@ -871,7 +871,7 @@ static int tls_session_destroy(void *session)
 
     pthread_mutex_lock(&ctx->mutex);
 
-    if (flb_socket_error(ptr->fd) == 0) {
+    if (ptr->fd >= 0 && flb_socket_error(ptr->fd) == 0) {
         SSL_shutdown(ptr->ssl);
     }
 
@@ -883,6 +883,32 @@ static int tls_session_destroy(void *session)
     return 0;
 }
 
+static void tls_session_invalidate(void *session)
+{
+    struct tls_session *ptr = session;
+    struct tls_context *ctx;
+
+    if (ptr == NULL) {
+        return;
+    }
+
+    ctx = ptr->parent;
+    if (ctx == NULL) {
+        ptr->fd = -1;
+        return;
+    }
+
+    pthread_mutex_lock(&ctx->mutex);
+
+    if (ptr->fd >= 0 && flb_socket_error(ptr->fd) == 0) {
+        SSL_shutdown(ptr->ssl);
+    }
+
+    ptr->fd = -1;
+
+    pthread_mutex_unlock(&ctx->mutex);
+}
+
 static const char *tls_session_alpn_get(void *session_)
 {
     const unsigned char    *backend_alpn_buffer;
@@ -1209,6 +1235,7 @@ static struct flb_tls_backend tls_openssl = {
     .set_minmax_proto     = tls_set_minmax_proto,
     .set_ciphers          = tls_set_ciphers,
     .session_create       = tls_session_create,
+    .session_invalidate   = tls_session_invalidate,
     .session_destroy      = tls_session_destroy,
     .net_read             = tls_net_read,
     .net_write            = tls_net_write,

tests/internal/CMakeLists.txt

@@ -55,6 +55,14 @@ set(UNIT_TESTS_FILES
   opentelemetry.c
 )
 
+# TLS helpers
+if(FLB_TLS)
+  set(UNIT_TESTS_FILES
+    ${UNIT_TESTS_FILES}
+    upstream_tls.c
+    )
+endif()
+
 # Config format
 set(UNIT_TESTS_FILES
     ${UNIT_TESTS_FILES}

tests/internal/upstream_tls.c

@@ -0,0 +1,104 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+#include <fluent-bit/flb_info.h>
+#include <fluent-bit/flb_upstream.h>
+#include <fluent-bit/flb_upstream_conn.h>
+#include <fluent-bit/flb_connection.h>
+#include <fluent-bit/flb_pipe.h>
+#include <fluent-bit/flb_socket.h>
+#include <fluent-bit/tls/flb_tls.h>
+
+#include "flb_tests_internal.h"
+
+#ifdef FLB_HAVE_TLS
+
+#ifdef FLB_SYSTEM_WINDOWS
+#include <fluent-bit/flb_compat.h>
+#endif
+
+struct test_backend_ctx {
+    int invalidate_calls;
+};
+
+static void test_session_invalidate(void *session)
+{
+    struct test_backend_ctx *ctx = session;
+
+    if (ctx != NULL) {
+        ctx->invalidate_calls++;
+    }
+}
+
+void test_prepare_destroy_conn_marks_tls_session_stale(void)
+{
+    struct test_backend_ctx backend_session = {0};
+    struct flb_tls_backend backend_api = {0};
+    struct flb_tls tls_context = {0};
+    struct flb_tls_session tls_session = {0};
+    struct flb_connection conn = {0};
+    struct flb_upstream upstream = {0};
+    struct flb_config config = {0};
+    struct flb_upstream_queue *queue;
+    flb_pipefd_t socket_pair[2];
+    int ret;
+
+#ifdef FLB_SYSTEM_WINDOWS
+    WSADATA wsa_data;
+
+    WSAStartup(0x0201, &wsa_data);
+#endif
+
+    ret = flb_pipe_create(socket_pair);
+    TEST_CHECK(ret == 0);
+
+    backend_api.session_invalidate = test_session_invalidate;
+    tls_context.api = &backend_api;
+
+    tls_session.ptr = &backend_session;
+    tls_session.tls = &tls_context;
+    tls_session.connection = &conn;
+
+    config.is_shutting_down = FLB_FALSE;
+    upstream.base.config = &config;
+    upstream.base.net.keepalive = FLB_FALSE;
+    upstream.tcp_host = "example";
+    upstream.tcp_port = 443;
+
+    flb_upstream_queue_init(&upstream.queue);
+
+    conn.fd = socket_pair[0];
+    conn.event.fd = conn.fd;
+    conn.event.status = 0;
+    conn.stream = (struct flb_stream *) &upstream;
+    conn.net = &upstream.base.net;
+    conn.tls_session = &tls_session;
+    conn.net_error = 0;
+
+    mk_list_init(&conn._head);
+    queue = &upstream.queue;
+    mk_list_add(&conn._head, &queue->busy_queue);
+
+    ret = flb_upstream_conn_release(&conn);
+    TEST_CHECK(ret == 0);
+
+    TEST_CHECK(backend_session.invalidate_calls == 1);
+    TEST_CHECK(conn.fd == -1);
+    TEST_CHECK(conn.event.fd == -1);
+    TEST_CHECK(mk_list_size(&queue->destroy_queue) == 1);
+    TEST_CHECK(conn.shutdown_flag == FLB_TRUE);
+
+    flb_pipe_close(socket_pair[1]);
+
+#ifdef FLB_SYSTEM_WINDOWS
+    WSACleanup();
+#endif
+}
+
+#endif
+
+TEST_LIST = {
+#ifdef FLB_HAVE_TLS
+    {"prepare_destroy_conn_marks_tls_session_stale", test_prepare_destroy_conn_marks_tls_session_stale},
+#endif
+    {0}
+};