Implement rfc6587 auto detection

lib/logproto/CMakeLists.txt

@@ -11,6 +11,7 @@ set(LOGPROTO_HEADERS
     logproto/logproto-server.h
     logproto/logproto-text-client.h
     logproto/logproto-text-server.h
+    logproto/logproto-auto-server.h
     PARENT_SCOPE)
 
 set(LOGPROTO_SOURCES
@@ -25,6 +26,7 @@ set(LOGPROTO_SOURCES
     logproto/logproto-server.c
     logproto/logproto-text-client.c
     logproto/logproto-text-server.c
+    logproto/logproto-auto-server.c
     PARENT_SCOPE)
 
 add_test_subdirectory(tests)

lib/logproto/Makefile.am

@@ -11,6 +11,7 @@ logprotoinclude_HEADERS = \
 	lib/logproto/logproto-framed-server.h	\
 	lib/logproto/logproto-text-client.h  \
 	lib/logproto/logproto-text-server.h	\
+	lib/logproto/logproto-auto-server.h	\
 	lib/logproto/logproto-multiline-server.h \
 	lib/logproto/logproto-record-server.h \
 	lib/logproto/logproto-builtins.h	\
@@ -25,6 +26,7 @@ logproto_sources = \
 	lib/logproto/logproto-framed-server.c	\
 	lib/logproto/logproto-text-client.c  \
 	lib/logproto/logproto-text-server.c	\
+	lib/logproto/logproto-auto-server.c	\
 	lib/logproto/logproto-multiline-server.c \
 	lib/logproto/logproto-record-server.c \
 	lib/logproto/logproto-builtins.c

lib/logproto/logproto-auto-server.c

@@ -0,0 +1,106 @@
+/*
+ * Copyright (c) 2024 Balázs Scheidler <[email protected]>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) any later version.
+ *
+ * This library 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
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; 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.
+ *
+ */
+#include "logproto-auto-server.h"
+#include "logproto-text-server.h"
+#include "logproto-framed-server.h"
+#include "messages.h"
+
+typedef struct _LogProtoAutoServer
+{
+  LogProtoServer super;
+} LogProtoAutoServer;
+
+static LogProtoServer *
+_construct_detected_proto(LogProtoAutoServer *self, const gchar *detect_buffer, gsize detect_buffer_len)
+{
+  LogTransport *transport = log_transport_stack_get_active(&self->super.transport_stack);
+
+  if (g_ascii_isdigit(detect_buffer[0]))
+    {
+      msg_debug("Auto-detected octet-counted-framing on RFC6587 connection, using framed protocol",
+                evt_tag_int("fd", transport->fd));
+      return log_proto_framed_server_new(NULL, self->super.options);
+    }
+  if (detect_buffer[0] == '<')
+    {
+      msg_debug("Auto-detected non-transparent-framing on RFC6587 connection, using simple text protocol",
+                evt_tag_int("fd", transport->fd));
+    }
+  else
+    {
+      msg_debug("Unable to detect framing on RFC6587 connection, falling back to simple text transport",
+                evt_tag_int("fd", transport->fd),
+                evt_tag_mem("detect_buffer", detect_buffer, detect_buffer_len));
+    }
+  return log_proto_text_server_new(NULL, self->super.options);
+}
+
+static LogProtoPrepareAction
+log_proto_auto_server_prepare(LogProtoServer *s, GIOCondition *cond, gint *timeout G_GNUC_UNUSED)
+{
+  LogProtoAutoServer *self = (LogProtoAutoServer *) s;
+  LogTransport *transport = log_transport_stack_get_active(&self->super.transport_stack);
+
+  *cond = transport->cond;
+  if (*cond == 0)
+    *cond = G_IO_IN;
+
+  return LPPA_POLL_IO;
+}
+
+static LogProtoStatus
+log_proto_auto_handshake(LogProtoServer *s, gboolean *handshake_finished, LogProtoServer **proto_replacement)
+{
+  LogProtoAutoServer *self = (LogProtoAutoServer *) s;
+  LogTransport *transport = log_transport_stack_get_active(&self->super.transport_stack);
+  gchar detect_buffer[8];
+  gboolean moved_forward;
+  gint rc;
+
+  rc = log_transport_read_ahead(transport, detect_buffer, sizeof(detect_buffer), &moved_forward);
+  if (rc == 0)
+    return LPS_EOF;
+  else if (rc < 0)
+    {
+      if (errno == EAGAIN)
+        return LPS_AGAIN;
+      return LPS_ERROR;
+    }
+
+  *proto_replacement = _construct_detected_proto(self, detect_buffer, rc);
+  return LPS_SUCCESS;
+}
+
+LogProtoServer *
+log_proto_auto_server_new(LogTransport *transport, const LogProtoServerOptions *options)
+{
+  LogProtoAutoServer *self = g_new0(LogProtoAutoServer, 1);
+
+  /* we are not using our own transport stack, transport is to be passed to
+   * the LogProto implementation once we finished with detection */
+
+  log_proto_server_init(&self->super, transport, options);
+  self->super.handshake = log_proto_auto_handshake;
+  self->super.prepare = log_proto_auto_server_prepare;
+  return &self->super;
+}

lib/logproto/logproto-auto-server.h

@@ -0,0 +1,30 @@
+/*
+ * Copyright (c) 2024 Balázs Scheidler <[email protected]>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) any later version.
+ *
+ * This library 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
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; 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.
+ *
+ */
+#ifndef LOGPROTO_AUTO_SERVER_H_INCLUDED
+#define LOGPROTO_AUTO_SERVER_H_INCLUDED
+
+#include "logproto-server.h"
+
+LogProtoServer *log_proto_auto_server_new(LogTransport *transport, const LogProtoServerOptions *options);
+
+#endif

lib/logproto/logproto-builtins.c

@@ -26,6 +26,7 @@
 #include "logproto-text-server.h"
 #include "logproto-framed-client.h"
 #include "logproto-framed-server.h"
+#include "logproto-auto-server.h"
 #include "plugin.h"
 #include "plugin-types.h"
 
@@ -39,6 +40,7 @@ DEFINE_LOG_PROTO_SERVER(log_proto_text);
 DEFINE_LOG_PROTO_SERVER(log_proto_text_with_nuls);
 DEFINE_LOG_PROTO_CLIENT(log_proto_framed);
 DEFINE_LOG_PROTO_SERVER(log_proto_framed);
+DEFINE_LOG_PROTO_SERVER(log_proto_auto);
 
 static Plugin framed_server_plugins[] =
 {
@@ -50,6 +52,7 @@ static Plugin framed_server_plugins[] =
   LOG_PROTO_SERVER_PLUGIN(log_proto_text_with_nuls, "text-with-nuls"),
   LOG_PROTO_CLIENT_PLUGIN(log_proto_framed, "framed"),
   LOG_PROTO_SERVER_PLUGIN(log_proto_framed, "framed"),
+  LOG_PROTO_SERVER_PLUGIN(log_proto_auto, "auto"),
 };
 
 void

lib/logproto/logproto-server.h

@@ -89,7 +89,7 @@ struct _LogProtoServer
   LogProtoStatus (*fetch)(LogProtoServer *s, const guchar **msg, gsize *msg_len, gboolean *may_read,
                           LogTransportAuxData *aux, Bookmark *bookmark);
   gboolean (*validate_options)(LogProtoServer *s);
-  LogProtoStatus (*handshake)(LogProtoServer *s, gboolean *handshake_finished);
+  LogProtoStatus (*handshake)(LogProtoServer *s, gboolean *handshake_finished, LogProtoServer **proto_replacement);
   void (*free_fn)(LogProtoServer *s);
 };
 
@@ -100,11 +100,19 @@ log_proto_server_validate_options(LogProtoServer *self)
 }
 
 static inline LogProtoStatus
-log_proto_server_handshake(LogProtoServer *s, gboolean *handshake_finished)
+log_proto_server_handshake(LogProtoServer *s, gboolean *handshake_finished, LogProtoServer **proto_replacement)
 {
   if (s->handshake)
     {
-      return s->handshake(s, handshake_finished);
+      LogProtoStatus status;
+
+      g_assert(*proto_replacement == NULL);
+      status = s->handshake(s, handshake_finished, proto_replacement);
+      if (*proto_replacement)
+        {
+          g_assert(status == LPS_SUCCESS || status == LPS_AGAIN);
+        }
+      return status;
     }
   *handshake_finished = TRUE;
   return LPS_SUCCESS;

lib/logproto/tests/CMakeLists.txt

@@ -5,6 +5,7 @@ set(TEST_LOGPROTO_SOURCES
   test-text-server.c
   test-dgram-server.c
   test-framed-server.c
+  test-auto-server.c
   test-indented-multiline-server.c
   test-regexp-multiline-server.c)
 

lib/logproto/tests/Makefile.am

@@ -17,6 +17,7 @@ lib_logproto_tests_test_logproto_SOURCES = 			\
 	lib/logproto/tests/test-text-server.c			\
 	lib/logproto/tests/test-dgram-server.c			\
 	lib/logproto/tests/test-framed-server.c			\
+	lib/logproto/tests/test-auto-server.c			\
 	lib/logproto/tests/test-indented-multiline-server.c	\
 	lib/logproto/tests/test-regexp-multiline-server.c
 

lib/logproto/tests/test-auto-server.c

@@ -0,0 +1,139 @@
+/*
+ * Copyright (c) 2012-2019 Balabit
+ * Copyright (c) 2012-2013 Balázs Scheidler
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) any later version.
+ *
+ * This library 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
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; 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.
+ *
+ */
+
+#include <criterion/criterion.h>
+#include "libtest/mock-transport.h"
+#include "libtest/proto_lib.h"
+#include "libtest/msg_parse_lib.h"
+
+#include "logproto/logproto-auto-server.h"
+
+#include <errno.h>
+
+Test(log_proto, test_log_proto_initial_framing_too_long)
+{
+  LogProtoServer *proto;
+
+  proto = log_proto_auto_server_new(
+            log_transport_mock_stream_new(
+              "100000000 too long\n", -1,
+              LTM_EOF),
+            get_inited_proto_server_options());
+
+  assert_proto_server_handshake_failure(&proto, LPS_SUCCESS);
+  assert_proto_server_fetch_failure(proto, LPS_ERROR, NULL);
+  log_proto_server_free(proto);
+}
+
+Test(log_proto, test_log_proto_error_in_initial_frame)
+{
+  LogProtoServer *proto;
+
+  proto = log_proto_auto_server_new(
+            log_transport_mock_stream_new(
+              LTM_INJECT_ERROR(EIO)),
+            get_inited_proto_server_options());
+
+  assert_proto_server_handshake_failure(&proto, LPS_ERROR);
+  log_proto_server_free(proto);
+}
+
+Test(log_proto, test_log_proto_auto_server_no_framing)
+{
+  LogProtoServer *proto;
+
+  proto = log_proto_auto_server_new(
+            log_transport_mock_stream_new(
+              "abcdefghijklmnopqstuvwxyz\n", -1,
+              "01234567\n", -1,
+              "01234567\0", 9,
+              "abcdef", -1,
+              LTM_EOF),
+            get_inited_proto_server_options());
+
+  assert_proto_server_handshake(&proto);
+  assert_proto_server_fetch(proto, "abcdefghijklmnopqstuvwxyz", -1);
+  assert_proto_server_fetch(proto, "01234567", -1);
+  assert_proto_server_fetch(proto, "01234567", 8);
+  assert_proto_server_fetch(proto, "abcdef", -1);
+  assert_proto_server_fetch_failure(proto, LPS_EOF, NULL);
+  log_proto_server_free(proto);
+}
+
+Test(log_proto, test_log_proto_auto_server_opening_bracket)
+{
+  LogProtoServer *proto;
+
+  proto = log_proto_auto_server_new(
+            log_transport_mock_stream_new(
+              "<55> abcdefghijklmnopqstuvwxyz\n", -1,
+              "01234567\n", -1,
+              "01234567\0", 9,
+              "abcdef", -1,
+              LTM_EOF),
+            get_inited_proto_server_options());
+
+  assert_proto_server_handshake(&proto);
+  assert_proto_server_fetch(proto, "<55> abcdefghijklmnopqstuvwxyz", -1);
+  assert_proto_server_fetch(proto, "01234567", -1);
+  assert_proto_server_fetch(proto, "01234567", 8);
+  assert_proto_server_fetch(proto, "abcdef", -1);
+  assert_proto_server_fetch_failure(proto, LPS_EOF, NULL);
+  log_proto_server_free(proto);
+}
+
+Test(log_proto, test_log_proto_auto_server_with_framing)
+{
+  LogProtoServer *proto;
+
+  proto = log_proto_auto_server_new(
+            log_transport_mock_stream_new(
+              "32 0123456789ABCDEF0123456789ABCDEF", -1,
+              "10 01234567\n\n", -1,
+              "10 01234567\0\0", 13,
+              /* utf8 */
+              "30 árvíztűrőtükörfúrógép", -1,
+              /* iso-8859-2 */
+              "21 \xe1\x72\x76\xed\x7a\x74\xfb\x72\xf5\x74\xfc\x6b\xf6\x72\x66\xfa"      /*  |árvíztűrőtükörfú| */
+              "\x72\xf3\x67\xe9\x70", -1,                                                /*  |rógép|            */
+              /* ucs4 */
+              "32 \x00\x00\x00\xe1\x00\x00\x00\x72\x00\x00\x00\x76\x00\x00\x00\xed"      /* |...á...r...v...í| */
+              "\x00\x00\x00\x7a\x00\x00\x00\x74\x00\x00\x01\x71\x00\x00\x00\x72", 35,    /* |...z...t...ű...r|  */
+              LTM_EOF),
+            get_inited_proto_server_options());
+
+  assert_proto_server_handshake(&proto);
+  assert_proto_server_fetch(proto, "0123456789ABCDEF0123456789ABCDEF", -1);
+  assert_proto_server_fetch(proto, "01234567\n\n", -1);
+  assert_proto_server_fetch(proto, "01234567\0\0", 10);
+  assert_proto_server_fetch(proto, "árvíztűrőtükörfúrógép", -1);
+  assert_proto_server_fetch(proto,
+                            "\xe1\x72\x76\xed\x7a\x74\xfb\x72\xf5\x74\xfc\x6b\xf6\x72\x66\xfa"        /*  |.rv.zt.r.t.k.rf.| */
+                            "\x72\xf3\x67\xe9\x70", -1);                                              /*  |r.g.p|            */
+  assert_proto_server_fetch(proto,
+                            "\x00\x00\x00\xe1\x00\x00\x00\x72\x00\x00\x00\x76\x00\x00\x00\xed"        /* |...á...r...v...í| */
+                            "\x00\x00\x00\x7a\x00\x00\x00\x74\x00\x00\x01\x71\x00\x00\x00\x72", 32);  /* |...z...t...q...r|  */
+  assert_proto_server_fetch_failure(proto, LPS_EOF, NULL);
+  log_proto_server_free(proto);
+}

lib/logreader.c

@@ -426,7 +426,16 @@ static inline gint
 log_reader_process_handshake(LogReader *self)
 {
   gboolean handshake_finished = FALSE;
-  LogProtoStatus status = log_proto_server_handshake(self->proto, &handshake_finished);
+  LogProtoServer *proto_replacement = NULL;
+  LogProtoStatus status = log_proto_server_handshake(self->proto, &handshake_finished, &proto_replacement);
+
+  if (proto_replacement)
+    {
+      g_assert(handshake_finished == FALSE);
+      log_transport_stack_move(&proto_replacement->transport_stack, &self->proto->transport_stack);
+      log_proto_server_free(self->proto);
+      self->proto = proto_replacement;
+    }
 
   switch (status)
     {