logproto: add support for auto-detecting haproxy handshakes

Signed-off-by: Balazs Scheidler <balazs.scheidler@axoflow.com>

lib/logproto/logproto-auto-server.c

@@ -25,6 +25,8 @@
 #include "logproto-framed-server.h"
 #include "messages.h"
 
+#include "transport/transport-haproxy.h"
+
 enum
 {
   LPAS_FAILURE,
@@ -37,6 +39,7 @@ typedef struct _LogProtoAutoServer
   LogProtoServer super;
 
   gboolean tls_detected;
+  gboolean haproxy_detected;
 } LogProtoAutoServer;
 
 static LogProtoServer *
@@ -118,6 +121,32 @@ _is_tls_client_hello(const gchar *buf, gsize buf_len)
   return LPAS_SUCCESS;
 }
 
+static gint
+_is_haproxy_header(const gchar *buf, gsize buf_len)
+{
+  const gchar proxy_v1_signature[] = { 0x50, 0x52, 0x4F, 0x58, 0x59, 0x20 };
+  const gchar proxy_v2_signature[] = { 0x0D, 0x0A, 0x0D, 0x0A, 0x00, 0x0D, 0x0A, 0x51, 0x55, 0x49, 0x54, 0x0A };
+
+  if (buf_len < sizeof(proxy_v1_signature))
+    return LPAS_NEED_MORE_DATA;
+
+  if (memcmp(buf, proxy_v1_signature, sizeof(proxy_v1_signature)) == 0)
+    return LPAS_SUCCESS;
+
+  if (buf_len < sizeof(proxy_v2_signature))
+    return LPAS_NEED_MORE_DATA;
+
+  if (memcmp(buf, proxy_v2_signature, sizeof(proxy_v2_signature)) != 0)
+    return LPAS_FAILURE;
+
+  if (buf_len < sizeof(proxy_v2_signature) + 1)
+    return LPAS_NEED_MORE_DATA;
+
+  if ((buf[sizeof(proxy_v2_signature)] & 0xf0) != 0x20)
+    return LPAS_FAILURE;
+  return LPAS_SUCCESS;
+}
+
 static LogProtoPrepareAction
 log_proto_auto_server_poll_prepare(LogProtoServer *s, GIOCondition *cond, gint *timeout G_GNUC_UNUSED)
 {
@@ -176,6 +205,38 @@ log_proto_auto_handshake(LogProtoServer *s, gboolean *handshake_finished, LogPro
           break;
         }
     }
+  if (!self->haproxy_detected)
+    {
+      switch (_is_haproxy_header(detect_buffer, rc))
+        {
+        case LPAS_NEED_MORE_DATA:
+          if (moved_forward)
+            return LPS_AGAIN;
+          break;
+        case LPAS_SUCCESS:
+          self->haproxy_detected = TRUE;
+
+          /* FIXME: make this a factory */
+          log_transport_stack_add_transport(&self->super.transport_stack, LOG_TRANSPORT_HAPROXY,
+                                            log_transport_haproxy_new(self->super.transport_stack.active_transport, self->super.transport_stack.active_transport));
+
+          /* this is a haproxy header */
+          if (log_transport_stack_switch(&self->super.transport_stack, LOG_TRANSPORT_HAPROXY))
+            {
+              msg_debug("HAProxy header detected, switching to haproxy");
+              return LPS_AGAIN;
+            }
+          else
+            {
+              msg_error("HAProxy header detected, but haproxy transport is not set up");
+              return LPS_ERROR;
+            }
+          break;
+        default:
+          break;
+
+        }
+    }
   *proto_replacement = _construct_detected_proto(self, detect_buffer, rc);
   return LPS_SUCCESS;
 }

lib/transport/transport-factory-tls.c

@@ -39,7 +39,7 @@ _construct_transport(const LogTransportFactory *s, LogTransportStack *stack)
 
   tls_session_set_verifier(tls_session, self->tls_verifier);
 
-  return log_transport_tls_new(tls_session, log_transport_stack_get_active(stack));
+  return log_transport_tls_new(tls_session, log_transport_stack_get_transport(stack, LOG_TRANSPORT_SOCKET));
 }
 
 void

modules/afsocket/transport-mapper-inet.c

@@ -134,7 +134,7 @@ transport_mapper_inet_setup_stack(TransportMapper *s, LogTransportStack *stack)
     {
       LogTransportIndex switch_to;
 
-      if (self->tls_context && !_is_tls_required(self))
+      if (self->tls_context)
         switch_to = LOG_TRANSPORT_TLS;
       else
         switch_to = LOG_TRANSPORT_SOCKET;