network(), syslog(): Fixed a potential crash for TLS destinations during reload

lib/logwriter.c

@@ -1741,6 +1741,12 @@ log_writer_steal_proto(LogWriter *self)
   return proto;
 }
 
+LogProtoClient *
+log_writer_get_proto(LogWriter *self)
+{
+  return self->proto;
+}
+
 
 /* run in the main thread in reaction to a log_writer_reopen to change
  * the destination LogProtoClient instance. It needs to be ran in the main

lib/logwriter.h

@@ -86,6 +86,7 @@ gboolean log_writer_has_pending_writes(LogWriter *self);
 gboolean log_writer_opened(LogWriter *self);
 void log_writer_reopen(LogWriter *self, LogProtoClient *proto);
 LogProtoClient *log_writer_steal_proto(LogWriter *self);
+LogProtoClient *log_writer_get_proto(LogWriter *self);
 void log_writer_set_queue(LogWriter *self, LogQueue *queue);
 LogQueue *log_writer_get_queue(LogWriter *s);
 LogWriter *log_writer_new(guint32 flags, GlobalConfig *cfg);

lib/transport/transport-adapter.c

@@ -27,7 +27,7 @@ gssize
 log_transport_adapter_read_method(LogTransport *s, gpointer buf, gsize buflen, LogTransportAuxData *aux)
 {
   LogTransportAdapter *self = (LogTransportAdapter *) s;
-  LogTransport *transport = log_transport_stack_get_transport(s->stack, self->base_index);
+  LogTransport *transport = log_transport_stack_get_or_create_transport(s->stack, self->base_index);
 
   return log_transport_read(transport, buf, buflen, aux);
 }
@@ -36,7 +36,7 @@ gssize
 log_transport_adapter_write_method(LogTransport *s, const gpointer buf, gsize count)
 {
   LogTransportAdapter *self = (LogTransportAdapter *) s;
-  LogTransport *transport = log_transport_stack_get_transport(s->stack, self->base_index);
+  LogTransport *transport = log_transport_stack_get_or_create_transport(s->stack, self->base_index);
 
   return log_transport_write(transport, buf, count);
 }
@@ -45,7 +45,7 @@ gssize
 log_transport_adapter_writev_method(LogTransport *s, struct iovec *iov, gint iov_count)
 {
   LogTransportAdapter *self = (LogTransportAdapter *) s;
-  LogTransport *transport = log_transport_stack_get_transport(s->stack, self->base_index);
+  LogTransport *transport = log_transport_stack_get_or_create_transport(s->stack, self->base_index);
 
   return log_transport_writev(transport, iov, iov_count);
 }

lib/transport/transport-stack.c

@@ -59,7 +59,7 @@ log_transport_stack_switch(LogTransportStack *self, gint index)
 {
   g_assert(index < LOG_TRANSPORT__MAX);
   LogTransport *active_transport = log_transport_stack_get_active(self);
-  LogTransport *requested_transport = log_transport_stack_get_transport(self, index);
+  LogTransport *requested_transport = log_transport_stack_get_or_create_transport(self, index);
 
   if (!requested_transport)
     return FALSE;

lib/transport/transport-stack.h

@@ -108,7 +108,7 @@ struct _LogTransportStack
 };
 
 static inline LogTransport *
-log_transport_stack_get_transport(LogTransportStack *self, gint index)
+log_transport_stack_get_or_create_transport(LogTransportStack *self, gint index)
 {
   g_assert(index < LOG_TRANSPORT__MAX);
 
@@ -127,7 +127,16 @@ log_transport_stack_get_transport(LogTransportStack *self, gint index)
 static inline LogTransport *
 log_transport_stack_get_active(LogTransportStack *self)
 {
-  return log_transport_stack_get_transport(self, self->active_transport);
+  // TODO - Change it to log_transport_stack_get_transport after checking call sites
+  return log_transport_stack_get_or_create_transport(self, self->active_transport);
+}
+
+static inline LogTransport *
+log_transport_stack_get_transport(LogTransportStack *self, gint index)
+{
+  g_assert(index < LOG_TRANSPORT__MAX);
+
+  return self->transports[index];
 }
 
 void log_transport_stack_add_factory(LogTransportStack *self, LogTransportFactory *);

lib/transport/transport-tls.c

@@ -237,6 +237,14 @@ log_transport_tls_write_method(LogTransport *s, const gpointer buf, gsize buflen
   return -1;
 }
 
+TLSSession *
+log_tansport_tls_get_session(LogTransport *s)
+{
+  LogTransportTLS *self = (LogTransportTLS *)s;
+
+  return self->tls_session;
+}
+
 
 static void log_transport_tls_free_method(LogTransport *s);
 

lib/transport/transport-tls.h

@@ -28,5 +28,6 @@
 #include "transport/tls-context.h"
 
 LogTransport *log_transport_tls_new(TLSSession *tls_session, gint fd);
+TLSSession *log_tansport_tls_get_session(LogTransport *s);
 
 #endif

modules/afsocket/afsocket-dest.c

@@ -37,13 +37,6 @@
 #include <sys/types.h>
 #include <sys/socket.h>
 
-typedef struct _ReloadStoreItem
-{
-  LogProtoClientFactory *proto_factory;
-  GSockAddr *dest_addr;
-  LogWriter *writer;
-} ReloadStoreItem;
-
 static ReloadStoreItem *
 _reload_store_item_new(AFSocketDestDriver *afsocket_dd)
 {

modules/afsocket/afsocket-dest.h

@@ -32,6 +32,13 @@
 
 #include <iv.h>
 
+typedef struct _ReloadStoreItem
+{
+  LogProtoClientFactory *proto_factory;
+  GSockAddr *dest_addr;
+  LogWriter *writer;
+} ReloadStoreItem;
+
 typedef struct _AFSocketDestDriver AFSocketDestDriver;
 
 struct _AFSocketDestDriver

tests/copyright/policy

@@ -277,6 +277,7 @@ tests/light/functional_tests/source_drivers/syslog_source/auto/test_auto_proto\.
 tests/light/src/syslog_ng_ctl/prometheus_stats_handler.py
 tests/light/src/syslog_ng_config/statements/template/template\.py
 tests/light/src/syslog_ng_config/statements/__init__\.py
+tests/light/functional_tests/destination_drivers/network_destination/test_kept_alive_tls_connection_doing_handshake_after_reload\.py
 modules/correlation/id-counter\.[ch]$
 modules/correlation/group-lines.h
 modules/xml/windows-eventlog-xml-parser\.h

tests/light/functional_tests/destination_drivers/network_destination/test_kept_alive_tls_connection_doing_handshake_after_reload.py

@@ -0,0 +1,47 @@
+#!/usr/bin/env python
+#############################################################################
+# Copyright (c) 2024 Axoflow
+#
+# This program is free software; you can redistribute it and/or modify it
+# under the terms of the GNU General Public License version 2 as published
+# by the Free Software Foundation, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
+#
+# As an additional exemption you are allowed to compile & link against the
+# OpenSSL libraries as published by the OpenSSL project. See the file
+# COPYING for details.
+#
+#############################################################################
+from src.common.file import copy_shared_file
+
+
+def test_kept_alive_tls_connection_doing_handshake_after_reload(config, syslog_ng, port_allocator, testcase_parameters):
+    ca = copy_shared_file(testcase_parameters, "valid-ca.crt")
+
+    network_source = config.create_network_source(port=port_allocator())
+    tls_network_destination = config.create_network_destination(
+        ip="localhost", port=port_allocator(), transport="tls",
+        keep_alive="yes",
+        tls={
+            "ca-file": ca,
+            "peer-verify": "yes",
+        },
+    )
+
+    config.create_logpath(statements=[network_source, tls_network_destination])
+
+    tls_network_destination.start_listener()
+    syslog_ng.start(config)
+
+    syslog_ng.reload(config)
+    network_source.write_log("test msg")
+
+    assert tls_network_destination.read_until_logs(["test msg"])

tests/light/shared_files/valid-ca.crt

@@ -0,0 +1,12 @@
+-----BEGIN CERTIFICATE-----
+MIIBrzCCAWGgAwIBAgIUcnaPg4elxLxduWuz285K04ySXm8wBQYDK2VwMEwxCzAJ
+BgNVBAYTAkhVMQswCQYDVQQIDAJCUDERMA8GA1UEBwwIQnVkYXBlc3QxEDAOBgNV
+BAoMB0F4b2Zsb3cxCzAJBgNVBAMMAkNBMCAXDTI0MTIxNDEyMDgxMVoYDzIxMjQx
+MTIwMTIwODExWjBMMQswCQYDVQQGEwJIVTELMAkGA1UECAwCQlAxETAPBgNVBAcM
+CEJ1ZGFwZXN0MRAwDgYDVQQKDAdBeG9mbG93MQswCQYDVQQDDAJDQTAqMAUGAytl
+cAMhABy7FT1AbxGnqrainAD583ToCDPgewE9KEhcOyoOjx+fo1MwUTAdBgNVHQ4E
+FgQUvb+QlzW+T0PrjRBlQ8xFaqzdFpYwHwYDVR0jBBgwFoAUvb+QlzW+T0PrjRBl
+Q8xFaqzdFpYwDwYDVR0TAQH/BAUwAwEB/zAFBgMrZXADQQA6zlO5N4zE1EveBX8p
+0qt1pszyettEDG6SOMGQsZmo0DvBo/d8IUN0pvmupP8ODCbRtBm2BFYNpwkaMnv5
+iHkH
+-----END CERTIFICATE-----

tests/light/shared_files/valid-ca.key

@@ -0,0 +1,3 @@
+-----BEGIN PRIVATE KEY-----
+MC4CAQAwBQYDK2VwBCIEIIJdHC0MTNkrdOUeEMsbdsjB8XFE18fsTW85Gi79tdfy
+-----END PRIVATE KEY-----

tests/light/shared_files/valid-localhost.crt

@@ -0,0 +1,10 @@
+-----BEGIN CERTIFICATE-----
+MIIBXDCCAQ4CFHPirqN81POlHqbX64hq/v7yULDPMAUGAytlcDBMMQswCQYDVQQG
+EwJIVTELMAkGA1UECAwCQlAxETAPBgNVBAcMCEJ1ZGFwZXN0MRAwDgYDVQQKDAdB
+eG9mbG93MQswCQYDVQQDDAJDQTAgFw0yNDEyMTQxMjA5NDNaGA8yMTI0MTEyMDEy
+MDk0M1owUzELMAkGA1UEBhMCSFUxCzAJBgNVBAgMAkJQMREwDwYDVQQHDAhCdWRh
+cGVzdDEQMA4GA1UECgwHQXhvZmxvdzESMBAGA1UEAwwJbG9jYWxob3N0MCowBQYD
+K2VwAyEAiDEFvid9kcpBbPgxCBaVWuj8YJZ3qZB1nBdsBuBMwEkwBQYDK2VwA0EA
+vuyKnc/DhVcZgn1BInCRMMDU/15kpgwoI6y36IO3ftCUmSVR79zn//YiotaeSBf9
+2HRhksjez10aoLLJvDErDg==
+-----END CERTIFICATE-----

tests/light/shared_files/valid-localhost.key

@@ -0,0 +1,3 @@
+-----BEGIN PRIVATE KEY-----
+MC4CAQAwBQYDK2VwBCIEIGfwI2Ogf0umSj1p5yOosbwOnDqhOOP9rolfsky9gytI
+-----END PRIVATE KEY-----

tests/light/src/common/file.py

@@ -45,6 +45,11 @@ def copy_shared_file(testcase_parameters, shared_file_name):
     return Path(Path.cwd(), shared_file_name)
 
 
+def get_shared_file(shared_file_name):
+    absolute_framework_dir = Path(__file__).parents[2].resolve()
+    return absolute_framework_dir / "shared_files" / shared_file_name
+
+
 def delete_session_file(shared_file_name):
     shared_file_name = Path(shared_file_name)
     shared_file_name.unlink()

tests/light/src/common/network.py

@@ -48,7 +48,10 @@ async def stop(self):
 
     async def close_client(self):
         """After a client connection is accepted, new connections will be rejected until this method is called"""
-        await self._client.close()
+        try:
+            await self._client.close()
+        except ConnectionResetError:  # remove this after fixing two way shutdown (SSL_shutdown) in AxoSyslog
+            logger.warning("Client closed the connection prematurely")
 
     async def _create_client(self):
         return self.Client()

tests/light/src/driver_io/network/network_io.py

@@ -21,13 +21,15 @@
 #
 #############################################################################
 import socket
+import ssl
 from enum import Enum
 from enum import IntEnum
 from pathlib import Path
 
 from src.common.asynchronous import BackgroundEventLoop
 from src.common.blocking import DEFAULT_TIMEOUT
 from src.common.file import File
+from src.common.file import get_shared_file
 from src.common.network import SingleConnectionTCPServer
 from src.common.network import UDPServer
 from src.common.random_id import get_unique_id
@@ -84,8 +86,12 @@ def construct(self, port, host=None, ip_proto_version=None):
             def _construct(server, reader_class):
                 return reader_class(server), server
 
+            tls = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+            tls.load_cert_chain(get_shared_file("valid-localhost.crt"), get_shared_file("valid-localhost.key"))
+
             transport_mapping = {
                 NetworkIO.Transport.TCP: lambda: _construct(SingleConnectionTCPServer(port, host, ip_proto_version), message_readers.SingleLineStreamReader),
+                NetworkIO.Transport.TLS: lambda: _construct(SingleConnectionTCPServer(port, host, ip_proto_version, tls), message_readers.SingleLineStreamReader),
                 NetworkIO.Transport.UDP: lambda: _construct(UDPServer(port, host, ip_proto_version), message_readers.DatagramReader),
             }
             return transport_mapping[self]()

tests/light/src/syslog_ng_config/statements/destinations/network_destination.py

@@ -28,6 +28,7 @@ def map_transport(transport):
     mapping = {
         "tcp": NetworkIO.Transport.TCP,
         "udp": NetworkIO.Transport.UDP,
+        "tls": NetworkIO.Transport.TLS,
     }
     transport = transport.replace("_", "-").replace("'", "").replace('"', "").lower()