Hi,
There is a new lot of patches for the spoa/modescurity contrib.
Thierry
On Wed, 19 Apr 2017 11:24:36 +0200
Thierry Fournier <thierry.fourn...@ozon.io> wrote:
>
> > On 19 Apr 2017, at 09:16, Aleksandar Lazic <al-hapr...@none.at> wrote:
> >
> >
> >
> > Am 19-04-2017 05:51, schrieb Willy Tarreau:
> >> On Tue, Apr 18, 2017 at 11:55:46PM +0200, Aleksandar Lazic wrote:
> >>> Why not reuse the upcoming http/2 format.
> >>> HTTP/2 is *easy* to parse and the implementations of servers are growing?
> >> Are you kidding ? I mean you want everyone to have to implement HPACK etc ?
> >
> > Well there are a currently lot of http/2 servers out there, but I
> > understand your concerns.
> >
> >>> I know this is still on the todo list but I think this is worth to go
> >>> instead to add another protocol.
> >>> Or we can take a look into grpc, which is on top of http/2 with
> >>> protobuffer.
> >>> http://www.grpc.io/
> >>> As I take more time to think in this direction why not add a grpc filter
> >>> like the spoa filter?
> >>> Opinions?
> >> Code generators are always a pain to deal with. It reminds me the old days
> >> when I was using rpcgen. And grpc doesn't provide a C implementation so
> >> that
> >> kills it :-)
> >
> > Yeah I have also seen that they prefer c++, c#, go and other languages but
> > not offer a c lib.
> > This is strange just because the core is written in c, but I don't need to
> > understand everything ;-)
> >
> >> Thierry told me he's going to implement a simple TLV-like list which will
> >> be much easier to implement and easy to code on both sides.
> >
> > @Thierry:
> > Ah something like netstrings ?
> > https://cr.yp.to/proto/netstrings.txt
>
> Hi thanks for the link. I do not like this encoding because it is not easy to
> parse. I must read ascii representation of numbers and convert-it. I prefer
> simple binary format, with length (encoding in the same way than the SPOE
> protocol) and value.
>
> Hi, the main goal is providing:
> - a format easy to decode
> - not adding lot of dependencies to haproxy
>
> So, after brainstorm, I propose:
>
> - remove all data already accessible via existing sample-fetches from the new
> sample-fetch introduced by my previous patch. These data are:
> method, path, query string, http-version and the body. All of these data
> are
> available in some sample fetches and it will be great to reuse it. Note
> that
> the only data which keep are the headers.
>
> - For the headers I propose two format: The first format is for general use.
> It
> is simply the header block in the same format that we received (with the
> final
> \r\n for detecting truncated header bloc)
>
> - For header I propose also a binary format like this:
> <number of headers>
> [
> <header name length><header name content>
> <header value length><header value content>
> ]…
> This format is easy to decode because each length is encoded with the SPOE
> numeric values encoding method, so it is always embedded in SPOE client.
>
> The SPOE message description will become
>
> spoe-message check-request
> uniqueid args method path query req.ver req.hdrs_bin req.body_size
> req.body
>
> uniqueid is a string which allow to do matching between the mod security logs
> and
> the haproxy logs. I propose the uniqueid, but this string can be
> another
> think like an header containing a uniqueid previously generated.
>
> req.hrds_bin is the header bloc in binary format.
>
> req.body_size is the advertised size of the body and it is used to determine
> if the
> body is full or truncated.
>
> Thierry
>
> >> I personally
> >> think that adding dependencies on tons of libs to implement simple stuff
> >> is more of a problem than an help even for implementors, because when you
> >> start to enter some dependency hell or version incompatibilities, it's a
> >> real pain, especially when it was done only to find how to implement
> >> just a small list. These things are useful if you want to implement heavy
> >> inter-application communications but it's not really the case here.
> >
> > I got you.
> >
> >> Cheers,
> >> Willy
> >
> > Regards
> > Aleks
>
>
>From 2759c1584f08f397ae11672fa11d260970033406 Mon Sep 17 00:00:00 2001
From: Thierry FOURNIER <thierry.fourn...@ozon.io>
Date: Sun, 9 Apr 2017 05:41:27 +0200
Subject: [PATCH 1/5] BUG/MINOR: change header-declared function to static
inline
When we include the header proto/spoe.h in other files in the same
project, the compilator claim that the symbol have multiple definitions:
src/flt_spoe.o: In function `spoe_encode_varint':
~/git/haproxy/include/proto/spoe.h:45: multiple definition of `spoe_encode_varint'
src/proto_http.o:~/git/haproxy/include/proto/spoe.h:45: first defined here
---
include/proto/spoe.h | 16 ++++++++--------
1 file changed, 8 insertions(+), 8 deletions(-)
diff --git a/include/proto/spoe.h b/include/proto/spoe.h
index 06fb52d..da19db1 100644
--- a/include/proto/spoe.h
+++ b/include/proto/spoe.h
@@ -39,7 +39,7 @@
*
* On success, it returns the number of written bytes and <*buf> is moved after
* the encoded value. Otherwise, it returns -1. */
-int
+static inline int
spoe_encode_varint(uint64_t i, char **buf, char *end)
{
unsigned char *p = (unsigned char *)*buf;
@@ -76,7 +76,7 @@ spoe_encode_varint(uint64_t i, char **buf, char *end)
* 'spoe_encode_varint' for details about varint.
* On success, it returns the number of read bytes and <*buf> is moved after the
* varint. Otherwise, it returns -1. */
-int
+static inline int
spoe_decode_varint(char **buf, char *end, uint64_t *i)
{
unsigned char *p = (unsigned char *)*buf;
@@ -109,7 +109,7 @@ spoe_decode_varint(char **buf, char *end, uint64_t *i)
* error is triggered.
* On success, it returns <len> and <*buf> is moved after the encoded value. If
* an error occurred, it returns -1. */
-int
+static inline int
spoe_encode_buffer(const char *str, size_t len, char **buf, char *end)
{
char *p = *buf;
@@ -137,7 +137,7 @@ spoe_encode_buffer(const char *str, size_t len, char **buf, char *end)
* 'spoe_encode_buffer', but if there is not enough space, it does not fail.
* On success, it returns the number of copied bytes and <*buf> is moved after
* the encoded value. If an error occured, it returns -1. */
-int
+static inline int
spoe_encode_frag_buffer(const char *str, size_t len, char **buf, char *end)
{
char *p = *buf;
@@ -166,7 +166,7 @@ spoe_encode_frag_buffer(const char *str, size_t len, char **buf, char *end)
* points on the first byte of the buffer.
* On success, it returns the buffer length and <*buf> is moved after the
* encoded buffer. Otherwise, it returns -1. */
-int
+static inline int
spoe_decode_buffer(char **buf, char *end, char **str, size_t *len)
{
char *p = *buf;
@@ -195,7 +195,7 @@ spoe_decode_buffer(char **buf, char *end, char **str, size_t *len)
* partially encoded. In this case, the offset <*off> is updated to known how
* many bytes has been encoded. If <*off> is zero at the end, it means that all
* data has been encoded. */
-int
+static inline int
spoe_encode_data(struct sample *smp, unsigned int *off, char **buf, char *end)
{
char *p = *buf;
@@ -318,7 +318,7 @@ spoe_encode_data(struct sample *smp, unsigned int *off, char **buf, char *end)
* - ipv6: 16 bytes
* - binary and string: a buffer prefixed by its size, a variable-length
* integer (see spoe_decode_buffer) */
-int
+static inline int
spoe_skip_data(char **buf, char *end)
{
char *str, *p = *buf;
@@ -366,7 +366,7 @@ spoe_skip_data(char **buf, char *end)
/* Decode a typed data and fill <smp>. If an error occurred, -1 is returned,
* otherwise the number of read bytes is returned and <*buf> is moved after the
* decoded data. See spoe_skip_data for details. */
-int
+static inline int
spoe_decode_data(char **buf, char *end, struct sample *smp)
{
char *str, *p = *buf;
--
1.7.10.4
>From ceb85a19df09ad03b1a73437f3a5a7faddec9365 Mon Sep 17 00:00:00 2001
From: Thierry FOURNIER <thierry.fourn...@ozon.io>
Date: Wed, 19 Apr 2017 11:49:44 +0200
Subject: [PATCH 2/5] REORG: spoe: move spoe_encode_varint /
spoe_decode_varint from spoe to common
These encoding functions does general stuff and can be used in
other context than spoe. This patch moves the function spoe_encode_varint
and spoe_decode_varint from spoe to common. It also remove the prefix spoe.
These functions will be used for encoding values in new binary sample fetch.
---
contrib/spoa_example/spoa.c | 22 +++++-----
include/common/standard.h | 77 +++++++++++++++++++++++++++++++++++
include/proto/spoe.h | 93 +++++--------------------------------------
src/flt_spoe.c | 20 +++++-----
4 files changed, 107 insertions(+), 105 deletions(-)
diff --git a/contrib/spoa_example/spoa.c b/contrib/spoa_example/spoa.c
index 7c6d504..8fb62e0 100644
--- a/contrib/spoa_example/spoa.c
+++ b/contrib/spoa_example/spoa.c
@@ -267,7 +267,7 @@ check_max_frame_size(struct spoe_frame *frame, char **buf, char *end)
(type & SPOE_DATA_T_MASK) != SPOE_DATA_T_UINT32 &&
(type & SPOE_DATA_T_MASK) != SPOE_DATA_T_UINT64)
return -1;
- if (spoe_decode_varint(&p, end, &sz) == -1)
+ if (decode_varint(&p, end, &sz) == -1)
return -1;
/* Keep the lower value */
@@ -453,7 +453,7 @@ check_discon_status_code(struct spoe_frame *frame, char **buf, char *end)
(type & SPOE_DATA_T_MASK) != SPOE_DATA_T_UINT32 &&
(type & SPOE_DATA_T_MASK) != SPOE_DATA_T_UINT64)
return -1;
- if (spoe_decode_varint(&p, end, &sz) == -1)
+ if (decode_varint(&p, end, &sz) == -1)
return -1;
frame->client->status_code = (unsigned int)sz;
@@ -709,9 +709,9 @@ handle_hanotify(struct spoe_frame *frame)
}
/* Read the stream-id and frame-id */
- if (spoe_decode_varint(&p, end, &stream_id) == -1)
+ if (decode_varint(&p, end, &stream_id) == -1)
goto ignore;
- if (spoe_decode_varint(&p, end, &frame_id) == -1)
+ if (decode_varint(&p, end, &frame_id) == -1)
goto ignore;
frame->stream_id = (unsigned int)stream_id;
@@ -765,9 +765,9 @@ handle_hafrag(struct spoe_frame *frame)
p+= 4;
/* Read the stream-id and frame-id */
- if (spoe_decode_varint(&p, end, &stream_id) == -1)
+ if (decode_varint(&p, end, &stream_id) == -1)
goto ignore;
- if (spoe_decode_varint(&p, end, &frame_id) == -1)
+ if (decode_varint(&p, end, &frame_id) == -1)
goto ignore;
if (frame->fragmented == false ||
@@ -842,7 +842,7 @@ prepare_agenthello(struct spoe_frame *frame)
/* "max-frame-size" K/V item */
spoe_encode_buffer("max-frame-size", 14, &p ,end);
*p++ = SPOE_DATA_T_UINT32;
- spoe_encode_varint(client->max_frame_size, &p, end);
+ encode_varint(client->max_frame_size, &p, end);
DEBUG(frame->worker, "<%lu> Agent maximum frame size : %u",
client->id, client->max_frame_size);
@@ -917,7 +917,7 @@ prepare_agentdicon(struct spoe_frame *frame)
/* "status-code" K/V item */
spoe_encode_buffer("status-code", 11, &p, end);
*p++ = SPOE_DATA_T_UINT32;
- spoe_encode_varint(client->status_code, &p, end);
+ encode_varint(client->status_code, &p, end);
DEBUG(frame->worker, "<%lu> Disconnect status code : %u",
client->id, client->status_code);
@@ -956,8 +956,8 @@ prepare_agentack(struct spoe_frame *frame)
p += 4;
/* Set stream-id and frame-id for ACK frames */
- spoe_encode_varint(frame->stream_id, &p, end);
- spoe_encode_varint(frame->frame_id, &p, end);
+ encode_varint(frame->stream_id, &p, end);
+ encode_varint(frame->frame_id, &p, end);
DEBUG(frame->worker, "STREAM-ID=%u - FRAME-ID=%u",
frame->stream_id, frame->frame_id);
@@ -1369,7 +1369,7 @@ process_frame_cb(evutil_socket_t fd, short events, void *arg)
*p++ = SPOE_SCOPE_SESS; /* Arg 1: the scope */
spoe_encode_buffer("ip_score", 8, &p, end); /* Arg 2: variable name */
*p++ = SPOE_DATA_T_UINT32;
- spoe_encode_varint(frame->ip_score, &p, end); /* Arg 3: variable value */
+ encode_varint(frame->ip_score, &p, end); /* Arg 3: variable value */
frame->len = (p - frame->buf);
}
write_frame(NULL, frame);
diff --git a/include/common/standard.h b/include/common/standard.h
index be719f7..9bfd1ae 100644
--- a/include/common/standard.h
+++ b/include/common/standard.h
@@ -205,6 +205,83 @@ static inline const char *LIM2A(unsigned long n, const char *alt)
return ret;
}
+/* Encode the integer <i> into a varint (variable-length integer). The encoded
+ * value is copied in <*buf>. Here is the encoding format:
+ *
+ * 0 <= X < 240 : 1 byte (7.875 bits) [ XXXX XXXX ]
+ * 240 <= X < 2288 : 2 bytes (11 bits) [ 1111 XXXX ] [ 0XXX XXXX ]
+ * 2288 <= X < 264432 : 3 bytes (18 bits) [ 1111 XXXX ] [ 1XXX XXXX ] [ 0XXX XXXX ]
+ * 264432 <= X < 33818864 : 4 bytes (25 bits) [ 1111 XXXX ] [ 1XXX XXXX ]*2 [ 0XXX XXXX ]
+ * 33818864 <= X < 4328786160 : 5 bytes (32 bits) [ 1111 XXXX ] [ 1XXX XXXX ]*3 [ 0XXX XXXX ]
+ * ...
+ *
+ * On success, it returns the number of written bytes and <*buf> is moved after
+ * the encoded value. Otherwise, it returns -1. */
+static inline int
+encode_varint(uint64_t i, char **buf, char *end)
+{
+ unsigned char *p = (unsigned char *)*buf;
+ int r;
+
+ if (p >= (unsigned char *)end)
+ return -1;
+
+ if (i < 240) {
+ *p++ = i;
+ *buf = (char *)p;
+ return 1;
+ }
+
+ *p++ = (unsigned char)i | 240;
+ i = (i - 240) >> 4;
+ while (i >= 128) {
+ if (p >= (unsigned char *)end)
+ return -1;
+ *p++ = (unsigned char)i | 128;
+ i = (i - 128) >> 7;
+ }
+
+ if (p >= (unsigned char *)end)
+ return -1;
+ *p++ = (unsigned char)i;
+
+ r = ((char *)p - *buf);
+ *buf = (char *)p;
+ return r;
+}
+
+/* Decode a varint from <*buf> and save the decoded value in <*i>. See
+ * 'spoe_encode_varint' for details about varint.
+ * On success, it returns the number of read bytes and <*buf> is moved after the
+ * varint. Otherwise, it returns -1. */
+static inline int
+decode_varint(char **buf, char *end, uint64_t *i)
+{
+ unsigned char *p = (unsigned char *)*buf;
+ int r;
+
+ if (p >= (unsigned char *)end)
+ return -1;
+
+ *i = *p++;
+ if (*i < 240) {
+ *buf = (char *)p;
+ return 1;
+ }
+
+ r = 4;
+ do {
+ if (p >= (unsigned char *)end)
+ return -1;
+ *i += (uint64_t)*p << r;
+ r += 7;
+ } while (*p++ >= 128);
+
+ r = ((char *)p - *buf);
+ *buf = (char *)p;
+ return r;
+}
+
/* returns a locally allocated string containing the quoted encoding of the
* input string. The output may be truncated to QSTR_SIZE chars, but it is
* guaranteed that the string will always be properly terminated. Quotes are
diff --git a/include/proto/spoe.h b/include/proto/spoe.h
index da19db1..1372a7d 100644
--- a/include/proto/spoe.h
+++ b/include/proto/spoe.h
@@ -22,88 +22,13 @@
#ifndef _PROTO_SPOE_H
#define _PROTO_SPOE_H
+#include <common/standard.h>
+
#include <types/spoe.h>
#include <proto/sample.h>
-/* Encode the integer <i> into a varint (variable-length integer). The encoded
- * value is copied in <*buf>. Here is the encoding format:
- *
- * 0 <= X < 240 : 1 byte (7.875 bits) [ XXXX XXXX ]
- * 240 <= X < 2288 : 2 bytes (11 bits) [ 1111 XXXX ] [ 0XXX XXXX ]
- * 2288 <= X < 264432 : 3 bytes (18 bits) [ 1111 XXXX ] [ 1XXX XXXX ] [ 0XXX XXXX ]
- * 264432 <= X < 33818864 : 4 bytes (25 bits) [ 1111 XXXX ] [ 1XXX XXXX ]*2 [ 0XXX XXXX ]
- * 33818864 <= X < 4328786160 : 5 bytes (32 bits) [ 1111 XXXX ] [ 1XXX XXXX ]*3 [ 0XXX XXXX ]
- * ...
- *
- * On success, it returns the number of written bytes and <*buf> is moved after
- * the encoded value. Otherwise, it returns -1. */
-static inline int
-spoe_encode_varint(uint64_t i, char **buf, char *end)
-{
- unsigned char *p = (unsigned char *)*buf;
- int r;
-
- if (p >= (unsigned char *)end)
- return -1;
-
- if (i < 240) {
- *p++ = i;
- *buf = (char *)p;
- return 1;
- }
-
- *p++ = (unsigned char)i | 240;
- i = (i - 240) >> 4;
- while (i >= 128) {
- if (p >= (unsigned char *)end)
- return -1;
- *p++ = (unsigned char)i | 128;
- i = (i - 128) >> 7;
- }
-
- if (p >= (unsigned char *)end)
- return -1;
- *p++ = (unsigned char)i;
-
- r = ((char *)p - *buf);
- *buf = (char *)p;
- return r;
-}
-
-/* Decode a varint from <*buf> and save the decoded value in <*i>. See
- * 'spoe_encode_varint' for details about varint.
- * On success, it returns the number of read bytes and <*buf> is moved after the
- * varint. Otherwise, it returns -1. */
-static inline int
-spoe_decode_varint(char **buf, char *end, uint64_t *i)
-{
- unsigned char *p = (unsigned char *)*buf;
- int r;
-
- if (p >= (unsigned char *)end)
- return -1;
-
- *i = *p++;
- if (*i < 240) {
- *buf = (char *)p;
- return 1;
- }
-
- r = 4;
- do {
- if (p >= (unsigned char *)end)
- return -1;
- *i += (uint64_t)*p << r;
- r += 7;
- } while (*p++ >= 128);
-
- r = ((char *)p - *buf);
- *buf = (char *)p;
- return r;
-}
-
/* Encode a buffer. Its length <len> is encoded as a varint, followed by a copy
* of <str>. It must have enough space in <*buf> to encode the buffer, else an
* error is triggered.
@@ -124,7 +49,7 @@ spoe_encode_buffer(const char *str, size_t len, char **buf, char *end)
return 0;
}
- ret = spoe_encode_varint(len, &p, end);
+ ret = encode_varint(len, &p, end);
if (ret == -1 || p + len > end)
return -1;
@@ -152,7 +77,7 @@ spoe_encode_frag_buffer(const char *str, size_t len, char **buf, char *end)
return 0;
}
- ret = spoe_encode_varint(len, &p, end);
+ ret = encode_varint(len, &p, end);
if (ret == -1 || p >= end)
return -1;
@@ -176,7 +101,7 @@ spoe_decode_buffer(char **buf, char *end, char **str, size_t *len)
*str = NULL;
*len = 0;
- ret = spoe_decode_varint(&p, end, &sz);
+ ret = decode_varint(&p, end, &sz);
if (ret == -1 || p + sz > end)
return -1;
@@ -217,7 +142,7 @@ spoe_encode_data(struct sample *smp, unsigned int *off, char **buf, char *end)
case SMP_T_SINT:
*p++ = SPOE_DATA_T_INT64;
- if (spoe_encode_varint(smp->data.u.sint, &p, end) == -1)
+ if (encode_varint(smp->data.u.sint, &p, end) == -1)
return -1;
break;
@@ -313,7 +238,7 @@ spoe_encode_data(struct sample *smp, unsigned int *off, char **buf, char *end)
*
* A types data is composed of a type (1 byte) and corresponding data:
* - boolean: non additional data (0 bytes)
- * - integers: a variable-length integer (see spoe_decode_varint)
+ * - integers: a variable-length integer (see decode_varint)
* - ipv4: 4 bytes
* - ipv6: 16 bytes
* - binary and string: a buffer prefixed by its size, a variable-length
@@ -337,7 +262,7 @@ spoe_skip_data(char **buf, char *end)
case SPOE_DATA_T_INT64:
case SPOE_DATA_T_UINT32:
case SPOE_DATA_T_UINT64:
- if (spoe_decode_varint(&p, end, &v) == -1)
+ if (decode_varint(&p, end, &v) == -1)
return -1;
break;
case SPOE_DATA_T_IPV4:
@@ -386,7 +311,7 @@ spoe_decode_data(char **buf, char *end, struct sample *smp)
case SPOE_DATA_T_INT64:
case SPOE_DATA_T_UINT32:
case SPOE_DATA_T_UINT64:
- if (spoe_decode_varint(&p, end, (uint64_t *)&smp->data.u.sint) == -1)
+ if (decode_varint(&p, end, (uint64_t *)&smp->data.u.sint) == -1)
return -1;
smp->data.type = SMP_T_SINT;
break;
diff --git a/src/flt_spoe.c b/src/flt_spoe.c
index 42e9c0c..9288de9 100644
--- a/src/flt_spoe.c
+++ b/src/flt_spoe.c
@@ -388,7 +388,7 @@ spoe_prepare_hahello_frame(struct appctx *appctx, char *frame, size_t size)
goto too_big;
*p++ = SPOE_DATA_T_UINT32;
- if (spoe_encode_varint(SPOE_APPCTX(appctx)->max_frame_size, &p, end) == -1)
+ if (encode_varint(SPOE_APPCTX(appctx)->max_frame_size, &p, end) == -1)
goto too_big;
/* "capabilities" K/V item */
@@ -469,7 +469,7 @@ spoe_prepare_hadiscon_frame(struct appctx *appctx, char *frame, size_t size)
goto too_big;
*p++ = SPOE_DATA_T_UINT32;
- if (spoe_encode_varint(SPOE_APPCTX(appctx)->status_code, &p, end) == -1)
+ if (encode_varint(SPOE_APPCTX(appctx)->status_code, &p, end) == -1)
goto too_big;
/* "message" K/V item */
@@ -527,9 +527,9 @@ spoe_prepare_hanotify_frame(struct appctx *appctx, struct spoe_context *ctx,
p += 4;
/* Set stream-id and frame-id */
- if (spoe_encode_varint(stream_id, &p, end) == -1)
+ if (encode_varint(stream_id, &p, end) == -1)
goto too_big;
- if (spoe_encode_varint(frame_id, &p, end) == -1)
+ if (encode_varint(frame_id, &p, end) == -1)
goto too_big;
/* Copy encoded messages, if possible */
@@ -583,9 +583,9 @@ spoe_prepare_hafrag_frame(struct appctx *appctx, struct spoe_context *ctx,
p += 4;
/* Set stream-id and frame-id */
- if (spoe_encode_varint(stream_id, &p, end) == -1)
+ if (encode_varint(stream_id, &p, end) == -1)
goto too_big;
- if (spoe_encode_varint(frame_id, &p, end) == -1)
+ if (encode_varint(frame_id, &p, end) == -1)
goto too_big;
if (ctx == NULL)
@@ -706,7 +706,7 @@ spoe_handle_agenthello_frame(struct appctx *appctx, char *frame, size_t size)
SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
return 0;
}
- if (spoe_decode_varint(&p, end, &sz) == -1) {
+ if (decode_varint(&p, end, &sz) == -1) {
SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
return 0;
}
@@ -858,7 +858,7 @@ spoe_handle_agentdiscon_frame(struct appctx *appctx, char *frame, size_t size)
SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
return 0;
}
- if (spoe_decode_varint(&p, end, &sz) == -1) {
+ if (decode_varint(&p, end, &sz) == -1) {
SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
return 0;
}
@@ -935,11 +935,11 @@ spoe_handle_agentack_frame(struct appctx *appctx, struct spoe_context **ctx,
}
/* Get the stream-id and the frame-id */
- if (spoe_decode_varint(&p, end, &stream_id) == -1) {
+ if (decode_varint(&p, end, &stream_id) == -1) {
SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
return 0;
}
- if (spoe_decode_varint(&p, end, &frame_id) == -1) {
+ if (decode_varint(&p, end, &frame_id) == -1) {
SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
return 0;
}
--
1.7.10.4
>From e843da078114860ea8d8bb4de48af183c7a95fc7 Mon Sep 17 00:00:00 2001
From: Thierry FOURNIER <thierry.fourn...@ozon.io>
Date: Sun, 9 Apr 2017 05:38:19 +0200
Subject: [PATCH 3/5] MINOR: Add binary encoding request header sample fetch
This sample fetch encodes the http request headers in binary
format. This sample-fetch is useful with SPOE.
---
doc/configuration.txt | 11 +++++
src/proto_http.c | 108 +++++++++++++++++++++++++++++++++++++++++++++++++
2 files changed, 119 insertions(+)
diff --git a/doc/configuration.txt b/doc/configuration.txt
index e6ea2cf..95dab55 100644
--- a/doc/configuration.txt
+++ b/doc/configuration.txt
@@ -14123,6 +14123,17 @@ payload_lv(<offset1>,<length>[,<offset2>]) : binary (deprecated)
(eg: "stick on", "stick match"), and for "res.payload_lv" when used in the
context of a response such as in "stick store response".
+req.hdrs_bin : binary
+ Returns the current request headers contained in preparsed binary form. This
+ is usefull for offloading some processing with SPOE. The format is easy to
+ understand. Each string is described by a length followed by a number of bytes
+ declared in the length.
+
+ <int:number-of-headers>*(<str:header-name><str:header-value>)
+
+ int: refers to the SPOE doc for understanding the encoding integers.
+ str: <int:length><bytes>
+
req.len : integer
req_len : integer (deprecated)
Returns an integer value corresponding to the number of bytes present in the
diff --git a/src/proto_http.c b/src/proto_http.c
index 24d034a..f689c7d 100644
--- a/src/proto_http.c
+++ b/src/proto_http.c
@@ -10435,6 +10435,112 @@ smp_fetch_uniqueid(const struct arg *args, struct sample *smp, const char *kw, v
return 1;
}
+/* Returns the header request in a length/value encoded format.
+ * This is useful for exchanges with the SPOE.
+ *
+ * A "length value" is a multibyte code encoding numbers. It uses the
+ * SPOE format. The encoding is the following:
+ *
+ * First, the length value is the number of couple "header name",
+ * "header value". Each couple "header name" / "header value" is composed
+ * like this:
+ * "length value" "header name bytes"
+ * "length value" "header value bytes"
+ */
+static int
+smp_fetch_hdrs_bin(const struct arg *args, struct sample *smp, const char *kw, void *private)
+{
+ struct http_msg *msg;
+ struct chunk *temp;
+ struct hdr_idx *idx;
+ const char *cur_ptr, *cur_next, *p;
+ int old_idx, cur_idx;
+ struct hdr_idx_elem *cur_hdr;
+ const char *hn, *hv;
+ int hnl, hvl;
+ int ret;
+ struct http_txn *txn;
+ char *buf;
+ char *end;
+
+ CHECK_HTTP_MESSAGE_FIRST();
+
+ temp = get_trash_chunk();
+ buf = temp->str;
+ end = temp->str + temp->size;
+
+ txn = smp->strm->txn;
+ idx = &txn->hdr_idx;
+ msg = &txn->req;
+
+ /* encode the number of headers. Note that the
+ * variable used take in account the header line.
+ */
+ ret = encode_varint(idx->used - 1, &buf, end);
+ if (ret == -1)
+ return 0;
+
+ /* Build array of headers. */
+ old_idx = 0;
+ cur_next = msg->chn->buf->p + hdr_idx_first_pos(idx);
+ while (1) {
+ cur_idx = idx->v[old_idx].next;
+ if (!cur_idx)
+ break;
+ old_idx = cur_idx;
+
+ cur_hdr = &idx->v[cur_idx];
+ cur_ptr = cur_next;
+ cur_next = cur_ptr + cur_hdr->len + cur_hdr->cr + 1;
+
+ /* Now we have one full header at cur_ptr of len cur_hdr->len,
+ * and the next header starts at cur_next. We'll check
+ * this header in the list as well as against the default
+ * rule.
+ */
+
+ /* look for ': *'. */
+ hn = cur_ptr;
+ for (p = cur_ptr; p < cur_ptr + cur_hdr->len && *p != ':'; p++);
+ if (p >= cur_ptr+cur_hdr->len)
+ continue;
+ hnl = p - hn;
+ p++;
+ while (p < cur_ptr + cur_hdr->len && (*p == ' ' || *p == '\t'))
+ p++;
+ if (p >= cur_ptr + cur_hdr->len)
+ continue;
+ hv = p;
+ hvl = cur_ptr + cur_hdr->len-p;
+
+ /* encode the header name. */
+ ret = encode_varint(hnl, &buf, end);
+ if (ret == -1)
+ return 0;
+ if (buf + hnl > end)
+ return 0;
+ memcpy(buf, hn, hnl);
+ buf += hnl;
+
+ /* encode and copy the value. */
+ ret = encode_varint(hvl, &buf, end);
+ if (ret == -1)
+ return 0;
+ if (buf + hvl > end)
+ return 0;
+ memcpy(buf, hv, hvl);
+ buf += hvl;
+ }
+
+ /* Initialise sample data which will be filled. */
+ smp->data.type = SMP_T_BIN;
+ smp->data.u.str.str = temp->str;
+ smp->data.u.str.len = buf - temp->str;
+ smp->data.u.str.size = temp->size;
+
+ return 1;
+}
+
/* returns the longest available part of the body. This requires that the body
* has been waited for using http-buffer-request.
*/
@@ -13345,6 +13451,8 @@ static struct sample_fetch_kw_list sample_fetch_keywords = {ILH, {
{ "req.body_size", smp_fetch_body_size, 0, NULL, SMP_T_SINT, SMP_USE_HRQHV },
{ "req.body_param", smp_fetch_body_param, ARG1(0,STR), NULL, SMP_T_BIN, SMP_USE_HRQHV },
+ { "req.hdrs_bin", smp_fetch_hdrs_bin, 0, NULL, SMP_T_BIN, SMP_USE_HRQHV },
+
/* HTTP version on the response path */
{ "res.ver", smp_fetch_stver, 0, NULL, SMP_T_STR, SMP_USE_HRSHV },
{ "resp_ver", smp_fetch_stver, 0, NULL, SMP_T_STR, SMP_USE_HRSHV },
--
1.7.10.4
>From 0bd3787b9373cb542bf253f5baed8ddfb1504b38 Mon Sep 17 00:00:00 2001
From: Thierry FOURNIER <thierry.fourn...@ozon.io>
Date: Wed, 19 Apr 2017 15:15:14 +0200
Subject: [PATCH 4/5] MINOR: proto-http: Add sample fetch wich returns all
HTTP headers
The sample fetch returns all headers including the last jump line.
The last jump line is used to determine if the block of headers is
truncated or not.
---
doc/configuration.txt | 6 ++++++
src/proto_http.c | 26 ++++++++++++++++++++++++++
2 files changed, 32 insertions(+)
diff --git a/doc/configuration.txt b/doc/configuration.txt
index 95dab55..37c7fda 100644
--- a/doc/configuration.txt
+++ b/doc/configuration.txt
@@ -14123,6 +14123,12 @@ payload_lv(<offset1>,<length>[,<offset2>]) : binary (deprecated)
(eg: "stick on", "stick match"), and for "res.payload_lv" when used in the
context of a response such as in "stick store response".
+req.hdrs : string
+ Returns the current request headers as string including the last empty line
+ separating headers from the request body. This empty can be used for detecting
+ truncated header bloc. This sample fetch is useful for some SPOE headers
+ analyzer.
+
req.hdrs_bin : binary
Returns the current request headers contained in preparsed binary form. This
is usefull for offloading some processing with SPOE. The format is easy to
diff --git a/src/proto_http.c b/src/proto_http.c
index f689c7d..0d27f9e 100644
--- a/src/proto_http.c
+++ b/src/proto_http.c
@@ -10435,6 +10435,31 @@ smp_fetch_uniqueid(const struct arg *args, struct sample *smp, const char *kw, v
return 1;
}
+/* Returns a string blok containing all headers including the
+ * empty ine wich separes headers from the body. This is useful
+ * form some headers analysis.
+ */
+static int
+smp_fetch_hdrs(const struct arg *args, struct sample *smp, const char *kw, void *private)
+{
+ struct http_msg *msg;
+ struct hdr_idx *idx;
+ struct http_txn *txn;
+
+ CHECK_HTTP_MESSAGE_FIRST();
+
+ txn = smp->strm->txn;
+ idx = &txn->hdr_idx;
+ msg = &txn->req;
+
+ smp->data.type = SMP_T_STR;
+ smp->data.u.str.str = msg->chn->buf->p + hdr_idx_first_pos(idx);
+ smp->data.u.str.len = msg->eoh - hdr_idx_first_pos(idx) + 1 +
+ (msg->chn->buf->p[msg->eoh] == '\r');
+
+ return 1;
+}
+
/* Returns the header request in a length/value encoded format.
* This is useful for exchanges with the SPOE.
*
@@ -13451,6 +13476,7 @@ static struct sample_fetch_kw_list sample_fetch_keywords = {ILH, {
{ "req.body_size", smp_fetch_body_size, 0, NULL, SMP_T_SINT, SMP_USE_HRQHV },
{ "req.body_param", smp_fetch_body_param, ARG1(0,STR), NULL, SMP_T_BIN, SMP_USE_HRQHV },
+ { "req.hdrs", smp_fetch_hdrs, 0, NULL, SMP_T_BIN, SMP_USE_HRQHV },
{ "req.hdrs_bin", smp_fetch_hdrs_bin, 0, NULL, SMP_T_BIN, SMP_USE_HRQHV },
/* HTTP version on the response path */
--
1.7.10.4
>From d3a138c177c6d7355dc24f79a1e61e86333869fd Mon Sep 17 00:00:00 2001
From: Thierry FOURNIER <thierry.fourn...@ozon.io>
Date: Mon, 10 Apr 2017 23:47:23 +0200
Subject: [PATCH 5/5] MINOR: Add ModSecurity wrapper as contrib
This patch contains a base for a modsecurity wrapper in HAProxy using SPOE.
---
contrib/modsecurity/Makefile | 45 +
contrib/modsecurity/README | 132 +++
contrib/modsecurity/modsec_wrapper.c | 621 +++++++++++
contrib/modsecurity/modsec_wrapper.h | 34 +
contrib/modsecurity/spoa.c | 1916 ++++++++++++++++++++++++++++++++++
contrib/modsecurity/spoa.h | 53 +
6 files changed, 2801 insertions(+)
create mode 100644 contrib/modsecurity/Makefile
create mode 100644 contrib/modsecurity/README
create mode 100644 contrib/modsecurity/modsec_wrapper.c
create mode 100644 contrib/modsecurity/modsec_wrapper.h
create mode 100644 contrib/modsecurity/spoa.c
create mode 100644 contrib/modsecurity/spoa.h
diff --git a/contrib/modsecurity/Makefile b/contrib/modsecurity/Makefile
new file mode 100644
index 0000000..72956f9
--- /dev/null
+++ b/contrib/modsecurity/Makefile
@@ -0,0 +1,45 @@
+DESTDIR =
+PREFIX = /usr/local
+BINDIR = $(PREFIX)/bin
+
+CC = gcc
+LD = $(CC)
+
+ifeq ($(MODSEC_INC),)
+MODSEC_INC := modsecurity-2.9.1/INSTALL/include
+endif
+
+ifeq ($(MODSEC_LIB),)
+MODSEC_LIB := modsecurity-2.9.1/INSTALL/lib
+endif
+
+ifeq ($(APACHE2_INC),)
+APACHE2_INC := /usr/include/apache2
+endif
+
+ifeq ($(APR_INC),)
+APR_INC := /usr/include/apr-1.0
+endif
+
+ifeq ($(LIBXML_INC),)
+LIBXML_INC := /usr/include/libxml2
+endif
+
+CFLAGS = -g -Wall -pthread
+LDFLAGS = -lpthread -levent -levent_pthreads -lcurl -lapr-1 -laprutil-1 -lxml2 -lpcre -lyajl
+INCS += -I../../include -I../../ebtree -I$(MODSEC_INC) -I$(APACHE2_INC) -I$(APR_INC) -I$(LIBXML_INC)
+LIBS =
+
+OBJS = spoa.o modsec_wrapper.o
+
+modsecurity: $(OBJS)
+ $(LD) $(LDFLAGS) $(LIBS) -o $@ $^ $(MODSEC_LIB)/standalone.a
+
+install: modsecurity
+ install modsecurity $(DESTDIR)$(BINDIR)
+
+clean:
+ rm -f modsecurity $(OBJS)
+
+%.o: %.c
+ $(CC) $(CFLAGS) $(INCS) -c -o $@ $<
diff --git a/contrib/modsecurity/README b/contrib/modsecurity/README
new file mode 100644
index 0000000..d4c9ecd
--- /dev/null
+++ b/contrib/modsecurity/README
@@ -0,0 +1,132 @@
+ModSecurity for HAProxy
+-----------------------
+
+This is a third party deamon whoch speaks SPOE. It give requests send by HAProxy
+to ModSecurity and returns the verdict.
+
+ Compilation
+---------------
+
+You must compile ModSecurity in standalone mode. Below an example for
+ModSecurity-2.9.1. Note that ModSecurity depends the Apache APR. I assume that
+the Apache dependencies are installed on the system.
+
+ ./configure \
+ --prefix=$PWD/INSTALL \
+ --disable-apache2-module \
+ --enable-standalone-module \
+ --enable-pcre-study \
+ --without-lua \
+ --enable-pcre-jit
+ make
+ make -C standalone install
+ mkdir -p $PWD/INSTALL/include
+ cp standalone/*.h $PWD/INSTALL/include
+ cp apache2/*.h $PWD/INSTALL/include
+
+Note that this compilation method works, but is a litle bit rustic. I cant
+deal with Lua, I supposed that is a dependecies problem on my computer.
+
+ Start the service
+---------------------
+
+After you have compiled it, to start the service, you just need to use "spoa"
+binary:
+
+ $> ./modsecurity -h
+ Usage: ./spoa [-h] [-d] [-p <port>] [-n <num-workers>] [-f <config-file>]
+ -h Print this message
+ -d Enable the debug mode
+ -f <config-file> Modsecurity configuration file
+ -m <max-frame-size> Specify the maximum frame size (default : 16384)
+ -p <port> Specify the port to listen on (default: 12345)
+ -n <num-workers> Specify the number of workers (default: 5)
+ -c <capability> Enable the support of the specified capability
+ -t <time> Set a delay to process a message (default: 0)
+ The value is specified in milliseconds by default,
+ but can be in any other unit if the number is suffixed
+ by a unit (us, ms, s)
+
+Note: A worker is a thread.
+
+
+ Configure a SPOE to use the service
+---------------------------------------
+
+All information about SPOE configuration can be found in "doc/SPOE.txt". Here is
+the configuration template to use for your SPOE with ModSecurity module:
+
+ [modsecurity]
+
+ spoe-agent modsecurity-agent
+ messages check-request
+ option var-prefix modsec
+ timeout hello 100ms
+ timeout idle 30s
+ timeout processing 15ms
+ use-backend spoe-modsecurity
+
+ spoe-message check-request
+ args unique-id method path query req.ver req.hdrs_bin req.body_size req.body
+ event on-frontend-http-request
+
+The engine is in the scope "modsecurity". So to enable it, you must set the
+following line in a frontend/listener section:
+
+ frontend my-front
+ ...
+ filter spoe engine modsecurity config spoe-modsecurity.conf
+ ...
+
+
+Because, in SPOE configuration file, we declare to use the backend
+"spoe-modsecurity" to communicate with the service, you must define it in
+HAProxy configuration. For example:
+
+ backend spoe-modsecurity
+ mode tcp
+ balance roundrobin
+ timeout connect 5s
+ timeout server 3m
+ server iprep1 127.0.0.1:12345
+
+The modsecurity action is returned in a variable called txn.modsec.code. It
+contains the HTTP returned code. If the variable contains 0, the request is
+clean.
+
+ tcp-request content reject if { var(txn.modsec.code) -m int gt 0 }
+
+With this rule, all the request not clean are reected.
+
+
+ Known bugs, limitations and TODO list
+-----------------------------------------
+
+Modsecurity bugs:
+-----------------
+
+* When the audit_log is used with the directive "SecAuditLogType Serial", in
+ some systems, the APR mutex initialisation silently fails, this causes a
+ segmentation fault. For my own usage, I have a patched version of modsec where
+ I use another mutex than "APR_LOCK_DEFAULT" like "APR_LOCK_PROC_PTHREAD"
+
+ - rc = apr_global_mutex_create(&msce->auditlog_lock, NULL, APR_LOCK_DEFAULT, mp);
+ + rc = apr_global_mutex_create(&msce->auditlog_lock, NULL, APR_LOCK_PROC_PTHREAD, mp);
+
+* Configuration file loaded with wilcard (eg. Include rules/*.conf), are loaded
+ in reverse alphabetical order. You can found a patch below. The ModSecurity
+ team ignored this patch.
+
+ https://github.com/SpiderLabs/ModSecurity/issues/1285
+ http://www.arpalert.org/0001-Fix-bug-when-load-files.patch
+
+ Or insert includes without wildcards.
+
+Todo:
+-----
+
+* Clarify the partial body analysis.
+* The response body is not yet analyzed.
+* ModSecurity can't modify the response body.
+* Implements real log management. Actually, the log are sent on stderr.
+* Implements daemon things (forks, write a pid, etc.).
diff --git a/contrib/modsecurity/modsec_wrapper.c b/contrib/modsecurity/modsec_wrapper.c
new file mode 100644
index 0000000..aad904b
--- /dev/null
+++ b/contrib/modsecurity/modsec_wrapper.c
@@ -0,0 +1,621 @@
+/*
+ * Modsecurity wrapper for haproxy
+ *
+ * This file contains the wrapper which sends data in ModSecurity
+ * and returns the verdict.
+ *
+ * Copyright 2016 OZON, Thierry Fournier <thierry.fourn...@ozon.io>
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License
+ * as published by the Free Software Foundation; either version
+ * 2 of the License, or (at your option) any later version.
+ *
+ */
+#include <limits.h>
+#include <stdio.h>
+#include <stdarg.h>
+
+#include <common/time.h>
+
+#include <types/global.h>
+#include <types/stream.h>
+
+#include <proto/arg.h>
+#include <proto/hdr_idx.h>
+#include <proto/hlua.h>
+#include <proto/log.h>
+#include <proto/proto_http.h>
+#include <proto/spoe.h>
+
+#include <api.h>
+
+#include "modsec_wrapper.h"
+#include "spoa.h"
+
+static char host_name[60];
+
+/* Note: The document and the code of "apr_table_make" considers
+ * that this function doesn't fails. The Apache APR code says
+ * other thing. If the system doesn't have any more memory, a
+ * a segfault occurs :(. Be carrefull with this module.
+ */
+
+struct directory_config *modsec_config = NULL;
+static server_rec *modsec_server = NULL;
+
+struct apr_bucket_haproxy {
+ apr_bucket_refcount refcount;
+ char *buffer;
+ size_t length;
+};
+
+static void haproxy_bucket_destroy(void *data)
+{
+ struct apr_bucket_haproxy *bucket = data;
+
+ if (apr_bucket_shared_destroy(bucket))
+ apr_bucket_free(bucket);
+}
+
+static apr_status_t haproxy_bucket_read(apr_bucket *bucket, const char **str,
+ apr_size_t *len, apr_read_type_e block)
+{
+ struct apr_bucket_haproxy *data = bucket->data;
+
+ if (bucket->start) {
+ *str = NULL;
+ *len = 0;
+ return APR_SUCCESS;
+ }
+
+ *str = data->buffer;
+ *len = data->length;
+ bucket->start = 1; /* Just a flag to say that the read is started */
+
+ return APR_SUCCESS;
+}
+
+static const apr_bucket_type_t apr_bucket_type_haproxy = {
+ "HAProxy", 7, APR_BUCKET_DATA,
+ haproxy_bucket_destroy,
+ haproxy_bucket_read,
+ apr_bucket_setaside_noop,
+ apr_bucket_shared_split,
+ apr_bucket_shared_copy
+};
+
+static char *chunk_strdup(struct request_rec *req, const char *str, size_t len)
+{
+ char *out;
+
+ out = apr_pcalloc(req->pool, len + 1);
+ if (!out)
+ return NULL;
+ memcpy(out, str, len);
+ out[len] = '\0';
+ return out;
+}
+
+static char *printf_dup(struct request_rec *req, char *fmt, ...)
+{
+ char *out;
+ va_list ap;
+ int len;
+
+ va_start(ap, fmt);
+ len = vsnprintf(NULL, 0, fmt, ap);
+ if (len == -1)
+ return NULL;
+ va_end(ap);
+
+ out = apr_pcalloc(req->pool, len + 1);
+ if (!out)
+ return NULL;
+
+ va_start(ap, fmt);
+ len = vsnprintf(out, len + 1, fmt, ap);
+ if (len == -1)
+ return NULL;
+ va_end(ap);
+
+ return out;
+}
+
+/* This function send logs. For now, it do nothing. */
+static void modsec_log(void *obj, int level, char *str)
+{
+ LOG(&null_worker, "%s", str);
+}
+
+/* This fucntion load the ModSecurity file. It returns -1 if the
+ * initialisation fails.
+ */
+int modsecurity_load(const char *file)
+{
+ const char *msg;
+ char cwd[128];
+
+ /* Initialises modsecurity. */
+
+ modsec_server = modsecInit();
+ if (modsec_server == NULL) {
+ LOG(&null_worker, "ModSecurity initilisation failed.\n");
+ return -1;
+ }
+
+ modsecSetLogHook(NULL, modsec_log);
+
+ gethostname(host_name, 60);
+ modsec_server->server_hostname = host_name;
+
+ modsecStartConfig();
+
+ modsec_config = modsecGetDefaultConfig();
+ if (modsec_config == NULL) {
+ LOG(&null_worker, "ModSecurity default configuration initilisation failed.\n");
+ return -1;
+ }
+
+ msg = modsecProcessConfig(modsec_config, file, getcwd(cwd, 128));
+ if (msg != NULL) {
+ LOG(&null_worker, "ModSecurity load configuration failed.\n");
+ return -1;
+ }
+
+ modsecFinalizeConfig();
+
+ modsecInitProcess();
+
+ return 1;
+}
+
+int modsecurity_process(struct worker *worker, struct modsecurity_parameters *params)
+{
+ struct conn_rec *cr;
+ struct request_rec *req;
+ struct apr_bucket_brigade *brigade;
+ struct apr_bucket *link_bucket;
+ struct apr_bucket_haproxy *data_bucket;
+ struct apr_bucket *last_bucket;
+ int i;
+ long clength;
+ char *err;
+ int fail;
+ const char *lang;
+ char *name, *value;
+ // int body_partial;
+ struct timeval now;
+ int ret;
+ char *buf;
+ char *end;
+ const char *uniqueid;
+ uint64_t uniqueid_len;
+ const char *meth;
+ uint64_t meth_len;
+ const char *path;
+ uint64_t path_len;
+ const char *qs;
+ uint64_t qs_len;
+ const char *vers;
+ uint64_t vers_len;
+ const char *body;
+ uint64_t body_len;
+ uint64_t body_exposed_len;
+ uint64_t hdr_nb;
+ struct {
+ const char *name;
+ uint64_t name_len;
+ const char *value;
+ uint64_t value_len;
+ } hdr[255];
+ int status;
+ int return_code = -1;
+
+ /* Decode uniqueid. */
+ uniqueid = params->uniqueid.data.u.str.str;
+ uniqueid_len = params->uniqueid.data.u.str.len;
+
+ /* Decode method. */
+ meth = params->method.data.u.str.str;
+ meth_len = params->method.data.u.str.len;
+
+ /* Decode path. */
+ path = params->path.data.u.str.str;
+ path_len = params->path.data.u.str.len;
+
+ /* Decode query string. */
+ qs = params->query.data.u.str.str;
+ qs_len = params->query.data.u.str.len;
+
+ /* Decode version. */
+ vers = params->vers.data.u.str.str;
+ vers_len = params->vers.data.u.str.len;
+
+ /* Decode header binary block. */
+ buf = params->hdrs_bin.data.u.str.str;
+ end = buf + params->hdrs_bin.data.u.str.len;
+
+ /* Decode number of headers. */
+ ret = decode_varint(&buf, end, &hdr_nb);
+ if (ret == -1)
+ return -1;
+ if (hdr_nb > 255)
+ return -1;
+
+ /* Decode each header. */
+ for (i = 0; i < hdr_nb; i++) {
+
+ /* Decode header name. */
+ ret = decode_varint(&buf, end, &hdr[i].name_len);
+ if (ret == -1)
+ return -1;
+ hdr[i].name = buf;
+ buf += hdr[i].name_len;
+ if (buf > end)
+ return -1;
+
+ /* Decode header value. */
+ ret = decode_varint(&buf, end, &hdr[i].value_len);
+ if (ret == -1)
+ return -1;
+ hdr[i].value = buf;
+ buf += hdr[i].value_len;
+ if (buf > end)
+ return -1;
+ }
+
+ /* Decode body length. */
+ body_exposed_len = (uint64_t)params->body_length.data.u.sint;
+
+ /* Decode body. */
+ body = params->body.data.u.str.str;
+ body_len = params->body.data.u.str.len;
+
+ fail = 1;
+
+ /* Init processing */
+
+ cr = modsecNewConnection();
+ req = modsecNewRequest(cr, modsec_config);
+
+ /* Load request. */
+
+ req->proxyreq = PROXYREQ_NONE;
+ req->header_only = 0; /* May modified later */
+
+ /* Copy header list. */
+
+ for (i = 0; i < hdr_nb; i++) {
+ name = chunk_strdup(req, hdr[i].name, hdr[i].name_len);
+ if (!name) {
+ errno = ENOMEM;
+ goto fail;
+ }
+ value = chunk_strdup(req, hdr[i].value, hdr[i].value_len);
+ if (!value) {
+ errno = ENOMEM;
+ goto fail;
+ }
+ apr_table_setn(req->headers_in, name, value);
+ }
+
+ /* Process special headers. */
+ req->range = apr_table_get(req->headers_in, "Range");
+ req->content_type = apr_table_get(req->headers_in, "Content-Type");
+ req->content_encoding = apr_table_get(req->headers_in, "Content-Encoding");
+ req->hostname = apr_table_get(req->headers_in, "Host");
+ req->parsed_uri.hostname = chunk_strdup(req, req->hostname, strlen(req->hostname));
+
+ lang = apr_table_get(req->headers_in, "Content-Languages");
+ if (lang != NULL) {
+ req->content_languages = apr_array_make(req->pool, 1, sizeof(const char *));
+ *(const char **)apr_array_push(req->content_languages) = lang;
+ }
+
+ lang = apr_table_get(req->headers_in, "Content-Length");
+ if (lang) {
+ errno = 0;
+ clength = strtol(lang, &err, 10);
+ if (*err != '\0' || errno != 0 || clength < 0 || clength > INT_MAX) {
+ errno = ERANGE;
+ goto fail;
+ }
+ req->clength = clength;
+ }
+
+ /* Copy the first line of the request. */
+ req->the_request = printf_dup(req, "%.*s %.*s%s%.*s %.*s",
+ meth_len, meth,
+ path_len, path,
+ qs_len > 0 ? "?" : "",
+ qs_len, qs,
+ vers_len, vers);
+ if (!req->the_request) {
+ errno = ENOMEM;
+ goto fail;
+ }
+
+ /* Copy the method. */
+ req->method = chunk_strdup(req, meth, meth_len);
+ if (!req->method) {
+ errno = ENOMEM;
+ goto fail;
+ }
+
+ /* Set the method number. */
+ if (meth_len < 3) {
+ errno = EINVAL;
+ goto fail;
+ }
+
+ /* Detect the method */
+ switch (meth_len) {
+ case 3:
+ if (strncmp(req->method, "GET", 3) == 0)
+ req->method_number = M_GET;
+ else if (strncmp(req->method, "PUT", 3) == 0)
+ req->method_number = M_PUT;
+ else {
+ errno = EINVAL;
+ goto fail;
+ }
+ break;
+ case 4:
+ if (strncmp(req->method, "POST", 4) == 0)
+ req->method_number = M_POST;
+ else if (strncmp(req->method, "HEAD", 4) == 0) {
+ req->method_number = M_GET;
+ req->header_only = 1;
+ }
+ else if (strncmp(req->method, "COPY", 4) == 0)
+ req->method_number = M_COPY;
+ else if (strncmp(req->method, "MOVE", 4) == 0)
+ req->method_number = M_MOVE;
+ else if (strncmp(req->method, "LOCK", 4) == 0)
+ req->method_number = M_LOCK;
+ else {
+ errno = EINVAL;
+ goto fail;
+ }
+ break;
+ case 5:
+ if (strncmp(req->method, "TRACE", 5) == 0)
+ req->method_number = M_TRACE;
+ else if (strncmp(req->method, "PATCH", 5) == 0)
+ req->method_number = M_PATCH;
+ else if (strncmp(req->method, "MKCOL", 5) == 0)
+ req->method_number = M_MKCOL;
+ else if (strncmp(req->method, "MERGE", 5) == 0)
+ req->method_number = M_MERGE;
+ else if (strncmp(req->method, "LABEL", 5) == 0)
+ req->method_number = M_LABEL;
+ else {
+ errno = EINVAL;
+ goto fail;
+ }
+ break;
+ case 6:
+ if (strncmp(req->method, "DELETE", 6) == 0)
+ req->method_number = M_DELETE;
+ else if (strncmp(req->method, "REPORT", 6) == 0)
+ req->method_number = M_REPORT;
+ else if (strncmp(req->method, "UPDATE", 6) == 0)
+ req->method_number = M_UPDATE;
+ else if (strncmp(req->method, "UNLOCK", 6) == 0)
+ req->method_number = M_UNLOCK;
+ else {
+ errno = EINVAL;
+ goto fail;
+ }
+ break;
+ case 7:
+ if (strncmp(req->method, "CHECKIN", 7) == 0)
+ req->method_number = M_CHECKIN;
+ else if (strncmp(req->method, "INVALID", 7) == 0)
+ req->method_number = M_INVALID;
+ else if (strncmp(req->method, "CONNECT", 7) == 0)
+ req->method_number = M_CONNECT;
+ else if (strncmp(req->method, "OPTIONS", 7) == 0)
+ req->method_number = M_OPTIONS;
+ else {
+ errno = EINVAL;
+ goto fail;
+ }
+ break;
+ case 8:
+ if (strncmp(req->method, "PROPFIND", 8) == 0)
+ req->method_number = M_PROPFIND;
+ else if (strncmp(req->method, "CHECKOUT", 8) == 0)
+ req->method_number = M_CHECKOUT;
+ else {
+ errno = EINVAL;
+ goto fail;
+ }
+ break;
+ case 9:
+ if (strncmp(req->method, "PROPPATCH", 9) == 0)
+ req->method_number = M_PROPPATCH;
+ else {
+ errno = EINVAL;
+ goto fail;
+ }
+ break;
+ case 10:
+ if (strncmp(req->method, "MKACTIVITY", 10) == 0)
+ req->method_number = M_MKACTIVITY;
+ else if (strncmp(req->method, "UNCHECKOUT", 10) == 0)
+ req->method_number = M_UNCHECKOUT;
+ else {
+ errno = EINVAL;
+ goto fail;
+ }
+ break;
+ case 11:
+ if (strncmp(req->method, "MKWORKSPACE", 11) == 0)
+ req->method_number = M_MKWORKSPACE;
+ else {
+ errno = EINVAL;
+ goto fail;
+ }
+ break;
+ case 15:
+ if (strncmp(req->method, "VERSION_CONTROL", 15) == 0)
+ req->method_number = M_VERSION_CONTROL;
+ else {
+ errno = EINVAL;
+ goto fail;
+ }
+ break;
+ case 16:
+ if (strncmp(req->method, "BASELINE_CONTROL", 16) == 0)
+ req->method_number = M_BASELINE_CONTROL;
+ else {
+ errno = EINVAL;
+ goto fail;
+ }
+ break;
+ default:
+ errno = EINVAL;
+ goto fail;
+ }
+
+ /* Copy the protocol. */
+ req->protocol = chunk_strdup(req, vers, vers_len);
+ if (!req->protocol) {
+ errno = ENOMEM;
+ goto fail;
+ }
+
+ /* Compute the protocol number. */
+ if (vers_len >= 8)
+ req->proto_num = 1000 + !!(vers[7] == '1');
+
+ /* The request time. */
+ gettimeofday(&now, NULL);
+ req->request_time = apr_time_make(now.tv_sec, now.tv_usec / 1000);
+
+ /* No status line. */
+ req->status_line = NULL;
+ req->status = 0;
+
+ /* Copy path. */
+ req->parsed_uri.path = chunk_strdup(req, path, path_len);
+ if (!req->parsed_uri.path) {
+ errno = ENOMEM;
+ goto fail;
+ }
+
+ /* Copy args (query string). */
+ req->args = chunk_strdup(req, qs, qs_len);
+ if (!req->args) {
+ errno = ENOMEM;
+ goto fail;
+ }
+
+ /* Set parsed_uri */
+
+ req->parsed_uri.scheme = "http";
+
+ if (req->hostname && req->parsed_uri.scheme && req->parsed_uri.path) {
+ i = snprintf(NULL, 0, "%s://%s%s",
+ req->parsed_uri.scheme, req->hostname, req->parsed_uri.path);
+ req->uri = apr_pcalloc(req->pool, i + 1);
+ if (!req->uri) {
+ errno = ENOMEM;
+ goto fail;
+ }
+ i = snprintf(req->uri, i + 1, "%s://%s%s",
+ req->parsed_uri.scheme, req->hostname, req->parsed_uri.path);
+ }
+
+ req->filename = req->parsed_uri.path;
+
+ /* Set unique id */
+
+ apr_table_setn(req->subprocess_env, "UNIQUE_ID", chunk_strdup(req, uniqueid, uniqueid_len));
+
+ /*
+ *
+ * Load body.
+ *
+ */
+
+ /* Create an empty bucket brigade */
+ brigade = apr_brigade_create(req->pool, req->connection->bucket_alloc);
+ if (!brigade) {
+ errno = ENOMEM;
+ goto fail;
+ }
+
+ /* Stores HTTP body avalaible data in a bucket */
+ data_bucket = apr_bucket_alloc(sizeof(*data_bucket), req->connection->bucket_alloc);
+ if (!data_bucket) {
+ errno = ENOMEM;
+ goto fail;
+ }
+ data_bucket->buffer = (char *)body;
+ data_bucket->length = body_len;
+
+ /* Create linked bucket */
+ link_bucket = apr_bucket_alloc(sizeof(*link_bucket), req->connection->bucket_alloc);
+ if (!link_bucket) {
+ errno = ENOMEM;
+ goto fail;
+ }
+ APR_BUCKET_INIT(link_bucket); /* link */
+ link_bucket->free = apr_bucket_free;
+ link_bucket->list = req->connection->bucket_alloc;
+ link_bucket = apr_bucket_shared_make(link_bucket, data_bucket, 0, body_len);
+ link_bucket->type = &apr_bucket_type_haproxy;
+
+ /* Insert the bucket at the end of the brigade. */
+ APR_BRIGADE_INSERT_TAIL(brigade, link_bucket);
+
+ /* Insert the last bucket. */
+ last_bucket = apr_bucket_eos_create(req->connection->bucket_alloc);
+ APR_BRIGADE_INSERT_TAIL(brigade, last_bucket);
+
+ /* Declares the bucket brigade in modsecurity */
+ modsecSetBodyBrigade(req, brigade);
+
+ /*
+ *
+ * Process analysis.
+ *
+ */
+
+ /* Process request headers analysis. */
+ status = modsecProcessRequestHeaders(req);
+ if (status != DECLINED && status != DONE)
+ return_code = status;
+
+ /* Process request body analysis. */
+ status = modsecProcessRequestBody(req);
+ if (status != DECLINED && status != DONE)
+ return_code = status;
+
+ /* End processing. */
+
+ fail = 0;
+ if (return_code == -1)
+ return_code = 0;
+
+fail:
+
+ modsecFinishRequest(req);
+ modsecFinishConnection(cr);
+
+ if (fail) {
+
+ /* errno == ERANGE / ENOMEM / EINVAL */
+ switch (errno) {
+ case ERANGE: LOG(worker, "Invalid range");
+ case ENOMEM: LOG(worker, "Out of memory error");
+ case EINVAL: LOG(worker, "Invalid value");
+ default: LOG(worker, "Unknown error");
+ }
+ }
+
+ return return_code;
+}
diff --git a/contrib/modsecurity/modsec_wrapper.h b/contrib/modsecurity/modsec_wrapper.h
new file mode 100644
index 0000000..e4b0484
--- /dev/null
+++ b/contrib/modsecurity/modsec_wrapper.h
@@ -0,0 +1,34 @@
+/*
+ * Modsecurity wrapper for haproxy
+ *
+ * This file contains the headers of the wrapper which sends data
+ * in ModSecurity and returns the verdict.
+ *
+ * Copyright 2016 OZON, Thierry Fournier <thierry.fourn...@ozon.io>
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License
+ * as published by the Free Software Foundation; either version
+ * 2 of the License, or (at your option) any later version.
+ *
+ */
+#ifndef __MODSEC_WRAPPER_H__
+#define __MODSEC_WRAPPER_H__
+
+#include "spoa.h"
+
+struct modsecurity_parameters {
+ struct sample uniqueid;
+ struct sample method;
+ struct sample path;
+ struct sample query;
+ struct sample vers;
+ struct sample hdrs_bin;
+ struct sample body_length;
+ struct sample body;
+};
+
+int modsecurity_load(const char *file);
+int modsecurity_process(struct worker *worker, struct modsecurity_parameters *params);
+
+#endif /* __MODSEC_WRAPPER_H__ */
diff --git a/contrib/modsecurity/spoa.c b/contrib/modsecurity/spoa.c
new file mode 100644
index 0000000..506ff82
--- /dev/null
+++ b/contrib/modsecurity/spoa.c
@@ -0,0 +1,1916 @@
+/*
+ * Modsecurity wrapper for haproxy
+ *
+ * This file contains the bootstrap for laucnching and scheduling modsecurity
+ * for working with HAProxy SPOE protocol.
+ *
+ * Copyright 2016 OZON, Thierry Fournier <thierry.fourn...@ozon.io>
+ *
+ * This file is inherited from "A Random IP reputation service acting as a Stream
+ * Processing Offload Agent"
+ *
+ * Copyright 2016 HAProxy Technologies, Christopher Faulet <cfau...@haproxy.com>
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License
+ * as published by the Free Software Foundation; either version
+ * 2 of the License, or (at your option) any later version.
+ *
+ */
+#include <unistd.h>
+#include <stdlib.h>
+#include <string.h>
+#include <stdbool.h>
+#include <errno.h>
+#include <stdio.h>
+#include <signal.h>
+#include <netinet/in.h>
+#include <sys/socket.h>
+#include <err.h>
+#include <ctype.h>
+
+#include <pthread.h>
+
+#include <event2/util.h>
+#include <event2/event.h>
+#include <event2/event_struct.h>
+#include <event2/thread.h>
+
+#include <common/mini-clist.h>
+#include <common/chunk.h>
+
+#include <proto/spoe.h>
+
+#include "spoa.h"
+#include "modsec_wrapper.h"
+
+#define DEFAULT_PORT 12345
+#define CONNECTION_BACKLOG 10
+#define NUM_WORKERS 10
+#define MAX_FRAME_SIZE 16384
+#define SPOP_VERSION "1.0"
+
+#define SLEN(str) (sizeof(str)-1)
+
+#define DEBUG(x...) \
+ do { \
+ if (debug) \
+ LOG(x); \
+ } while (0)
+
+
+enum spoa_state {
+ SPOA_ST_CONNECTING = 0,
+ SPOA_ST_PROCESSING,
+ SPOA_ST_DISCONNECTING,
+};
+
+enum spoa_frame_type {
+ SPOA_FRM_T_UNKNOWN = 0,
+ SPOA_FRM_T_HAPROXY,
+ SPOA_FRM_T_AGENT,
+};
+
+struct spoe_engine {
+ char *id;
+
+ struct list processing_frames;
+ struct list outgoing_frames;
+
+ struct list clients;
+ struct list list;
+};
+
+struct spoe_frame {
+ enum spoa_frame_type type;
+ char *buf;
+ unsigned int offset;
+ unsigned int len;
+
+ unsigned int stream_id;
+ unsigned int frame_id;
+ unsigned int flags;
+ bool hcheck; /* true is the CONNECT frame is a healthcheck */
+ bool fragmented; /* true if the frame is fragmented */
+ int modsec_code; /* modsecurity return code. -1 if unset, 0 if none, other it returns http code. */
+
+ struct event process_frame_event;
+ struct worker *worker;
+ struct spoe_engine *engine;
+ struct client *client;
+ struct list list;
+
+ char *frag_buf; /* used to accumulate payload of a fragmented frame */
+ unsigned int frag_len;
+
+ char data[0];
+};
+
+struct client {
+ int fd;
+ unsigned long id;
+ enum spoa_state state;
+
+ struct event read_frame_event;
+ struct event write_frame_event;
+
+ struct spoe_frame *incoming_frame;
+ struct spoe_frame *outgoing_frame;
+
+ struct list processing_frames;
+ struct list outgoing_frames;
+
+ unsigned int max_frame_size;
+ int status_code;
+
+ char *engine_id;
+ struct spoe_engine *engine;
+ bool pipelining;
+ bool async;
+ bool fragmentation;
+
+ struct worker *worker;
+ struct list by_worker;
+ struct list by_engine;
+};
+
+/* Globals */
+static struct worker *workers = NULL;
+ struct worker null_worker = { .id = 0 };
+static unsigned long clicount = 0;
+static int server_port = DEFAULT_PORT;
+static int num_workers = NUM_WORKERS;
+static unsigned int max_frame_size = MAX_FRAME_SIZE;
+struct timeval processing_delay = {0, 0};
+static bool debug = false;
+static bool pipelining = false;
+static bool async = false;
+static bool fragmentation = false;
+
+
+static const char *spoe_frm_err_reasons[SPOE_FRM_ERRS] = {
+ [SPOE_FRM_ERR_NONE] = "normal",
+ [SPOE_FRM_ERR_IO] = "I/O error",
+ [SPOE_FRM_ERR_TOUT] = "a timeout occurred",
+ [SPOE_FRM_ERR_TOO_BIG] = "frame is too big",
+ [SPOE_FRM_ERR_INVALID] = "invalid frame received",
+ [SPOE_FRM_ERR_NO_VSN] = "version value not found",
+ [SPOE_FRM_ERR_NO_FRAME_SIZE] = "max-frame-size value not found",
+ [SPOE_FRM_ERR_NO_CAP] = "capabilities value not found",
+ [SPOE_FRM_ERR_BAD_VSN] = "unsupported version",
+ [SPOE_FRM_ERR_BAD_FRAME_SIZE] = "max-frame-size too big or too small",
+ [SPOE_FRM_ERR_FRAG_NOT_SUPPORTED] = "fragmentation not supported",
+ [SPOE_FRM_ERR_INTERLACED_FRAMES] = "invalid interlaced frames",
+ [SPOE_FRM_ERR_FRAMEID_NOTFOUND] = "frame-id not found",
+ [SPOE_FRM_ERR_RES] = "resource allocation error",
+ [SPOE_FRM_ERR_UNKNOWN] = "an unknown error occurred",
+};
+
+static void signal_cb(evutil_socket_t, short, void *);
+static void accept_cb(evutil_socket_t, short, void *);
+static void worker_monitor_cb(evutil_socket_t, short, void *);
+static void process_frame_cb(evutil_socket_t, short, void *);
+static void read_frame_cb(evutil_socket_t, short, void *);
+static void write_frame_cb(evutil_socket_t, short, void *);
+
+static void use_spoe_engine(struct client *);
+static void unuse_spoe_engine(struct client *);
+static void release_frame(struct spoe_frame *);
+static void release_client(struct client *);
+
+/* Check the protocol version. It returns -1 if an error occurred, the number of
+ * read bytes otherwise. */
+static int
+check_proto_version(struct spoe_frame *frame, char **buf, char *end)
+{
+ char *str, *p = *buf;
+ uint64_t sz;
+ int ret;
+
+ /* Get the list of all supported versions by HAProxy */
+ if ((*p++ & SPOE_DATA_T_MASK) != SPOE_DATA_T_STR)
+ return -1;
+ ret = spoe_decode_buffer(&p, end, &str, &sz);
+ if (ret == -1 || !str)
+ return -1;
+
+ DEBUG(frame->worker, "<%lu> Supported versions : %.*s",
+ frame->client->id, (int)sz, str);
+
+ /* TODO: Find the right verion in supported ones */
+
+ ret = (p - *buf);
+ *buf = p;
+ return ret;
+}
+
+/* Check max frame size value. It returns -1 if an error occurred, the number of
+ * read bytes otherwise. */
+static int
+check_max_frame_size(struct spoe_frame *frame, char **buf, char *end)
+{
+ char *p = *buf;
+ uint64_t sz;
+ int type, ret;
+
+ /* Get the max-frame-size value of HAProxy */
+ type = *p++;
+ if ((type & SPOE_DATA_T_MASK) != SPOE_DATA_T_INT32 &&
+ (type & SPOE_DATA_T_MASK) != SPOE_DATA_T_INT64 &&
+ (type & SPOE_DATA_T_MASK) != SPOE_DATA_T_UINT32 &&
+ (type & SPOE_DATA_T_MASK) != SPOE_DATA_T_UINT64)
+ return -1;
+ if (decode_varint(&p, end, &sz) == -1)
+ return -1;
+
+ /* Keep the lower value */
+ if (sz < frame->client->max_frame_size)
+ frame->client->max_frame_size = sz;
+
+ DEBUG(frame->worker, "<%lu> HAProxy maximum frame size : %u",
+ frame->client->id, (unsigned int)sz);
+
+ ret = (p - *buf);
+ *buf = p;
+ return ret;
+}
+
+/* Check healthcheck value. It returns -1 if an error occurred, the number of
+ * read bytes otherwise. */
+static int
+check_healthcheck(struct spoe_frame *frame, char **buf, char *end)
+{
+ char *p = *buf;
+ int type, ret;
+
+ /* Get the "healthcheck" value */
+ type = *p++;
+ if ((type & SPOE_DATA_T_MASK) != SPOE_DATA_T_BOOL)
+ return -1;
+ frame->hcheck = ((type & SPOE_DATA_FL_TRUE) == SPOE_DATA_FL_TRUE);
+
+ DEBUG(frame->worker, "<%lu> HELLO healthcheck : %s",
+ frame->client->id, (frame->hcheck ? "true" : "false"));
+
+ ret = (p - *buf);
+ *buf = p;
+ return ret;
+}
+
+/* Check capabilities value. It returns -1 if an error occurred, the number of
+ * read bytes otherwise. */
+static int
+check_capabilities(struct spoe_frame *frame, char **buf, char *end)
+{
+ struct client *client = frame->client;
+ char *str, *p = *buf;
+ uint64_t sz;
+ int ret;
+
+ if ((*p++ & SPOE_DATA_T_MASK) != SPOE_DATA_T_STR)
+ return -1;
+ if (spoe_decode_buffer(&p, end, &str, &sz) == -1)
+ return -1;
+ if (str == NULL) /* this is not an error */
+ goto end;
+
+ DEBUG(frame->worker, "<%lu> HAProxy capabilities : %.*s",
+ client->id, (int)sz, str);
+
+ while (sz) {
+ char *delim;
+
+ /* Skip leading spaces */
+ for (; isspace(*str) && sz; sz--);
+
+ if (sz >= 10 && !strncmp(str, "pipelining", 10)) {
+ str += 10; sz -= 10;
+ if (!sz || isspace(*str) || *str == ',') {
+ DEBUG(frame->worker,
+ "<%lu> HAProxy supports frame pipelining",
+ client->id);
+ client->pipelining = true;
+ }
+ }
+ else if (sz >= 5 && !strncmp(str, "async", 5)) {
+ str += 5; sz -= 5;
+ if (!sz || isspace(*str) || *str == ',') {
+ DEBUG(frame->worker,
+ "<%lu> HAProxy supports asynchronous frame",
+ client->id);
+ client->async = true;
+ }
+ }
+ else if (sz >= 13 && !strncmp(str, "fragmentation", 13)) {
+ str += 13; sz -= 13;
+ if (!sz || isspace(*str) || *str == ',') {
+ DEBUG(frame->worker,
+ "<%lu> HAProxy supports fragmented frame",
+ client->id);
+ client->fragmentation = true;
+ }
+ }
+
+ if (!sz || (delim = memchr(str, ',', sz)) == NULL)
+ break;
+ delim++;
+ sz -= (delim - str);
+ str = delim;
+ }
+ end:
+ ret = (p - *buf);
+ *buf = p;
+ return ret;
+}
+
+/* Check engine-id value. It returns -1 if an error occurred, the number of
+ * read bytes otherwise. */
+static int
+check_engine_id(struct spoe_frame *frame, char **buf, char *end)
+{
+ struct client *client = frame->client;
+ char *str, *p = *buf;
+ uint64_t sz;
+ int ret;
+
+ if ((*p++ & SPOE_DATA_T_MASK) != SPOE_DATA_T_STR)
+ return -1;
+
+ if (spoe_decode_buffer(&p, end, &str, &sz) == -1)
+ return -1;
+ if (str == NULL) /* this is not an error */
+ goto end;
+
+ if (client->engine != NULL)
+ goto end;
+
+ DEBUG(frame->worker, "<%lu> HAProxy engine id : %.*s",
+ client->id, (int)sz, str);
+
+ client->engine_id = strndup(str, (int)sz);
+ end:
+ ret = (p - *buf);
+ *buf = p;
+ return ret;
+}
+
+static int
+acc_payload(struct spoe_frame *frame)
+{
+ struct client *client = frame->client;
+ char *buf;
+ size_t len = frame->len - frame->offset;
+ int ret = frame->offset;
+
+ /* No need to accumulation payload */
+ if (frame->fragmented == false)
+ return ret;
+
+ buf = realloc(frame->frag_buf, frame->frag_len + len);
+ if (buf == NULL) {
+ client->status_code = SPOE_FRM_ERR_RES;
+ return -1;
+ }
+ memcpy(buf + frame->frag_len, frame->buf + frame->offset, len);
+ frame->frag_buf = buf;
+ frame->frag_len += len;
+
+ if (!(frame->flags & SPOE_FRM_FL_FIN)) {
+ /* Wait for next parts */
+ frame->buf = (char *)(frame->data);
+ frame->offset = 0;
+ frame->len = 0;
+ frame->flags = 0;
+ return 1;
+ }
+
+ frame->buf = frame->frag_buf;
+ frame->len = frame->frag_len;
+ frame->offset = 0;
+ return ret;
+}
+
+/* Check disconnect status code. It returns -1 if an error occurred, the number
+ * of read bytes otherwise. */
+static int
+check_discon_status_code(struct spoe_frame *frame, char **buf, char *end)
+{
+ char *p = *buf;
+ uint64_t sz;
+ int type, ret;
+
+ /* Get the "status-code" value */
+ type = *p++;
+ if ((type & SPOE_DATA_T_MASK) != SPOE_DATA_T_INT32 &&
+ (type & SPOE_DATA_T_MASK) != SPOE_DATA_T_INT64 &&
+ (type & SPOE_DATA_T_MASK) != SPOE_DATA_T_UINT32 &&
+ (type & SPOE_DATA_T_MASK) != SPOE_DATA_T_UINT64)
+ return -1;
+ if (decode_varint(&p, end, &sz) == -1)
+ return -1;
+
+ frame->client->status_code = (unsigned int)sz;
+
+ DEBUG(frame->worker, "<%lu> Disconnect status code : %u",
+ frame->client->id, frame->client->status_code);
+
+ ret = (p - *buf);
+ *buf = p;
+ return ret;
+}
+
+/* Check the disconnect message. It returns -1 if an error occurred, the number
+ * of read bytes otherwise. */
+static int
+check_discon_message(struct spoe_frame *frame, char **buf, char *end)
+{
+ char *str, *p = *buf;
+ uint64_t sz;
+ int ret;
+
+ /* Get the "message" value */
+ if ((*p++ & SPOE_DATA_T_MASK) != SPOE_DATA_T_STR)
+ return -1;
+ ret = spoe_decode_buffer(&p, end, &str, &sz);
+ if (ret == -1 || !str)
+ return -1;
+
+ DEBUG(frame->worker, "<%lu> Disconnect message : %.*s",
+ frame->client->id, (int)sz, str);
+
+ ret = (p - *buf);
+ *buf = p;
+ return ret;
+}
+
+
+
+/* Decode a HELLO frame received from HAProxy. It returns -1 if an error
+ * occurred, otherwise the number of read bytes. HELLO frame cannot be
+ * ignored and having another frame than a HELLO frame is an error. */
+static int
+handle_hahello(struct spoe_frame *frame)
+{
+ struct client *client = frame->client;
+ char *p, *end;
+
+ p = frame->buf;
+ end = frame->buf + frame->len;
+
+ /* Check frame type: we really want a HELLO frame */
+ if (*p++ != SPOE_FRM_T_HAPROXY_HELLO)
+ goto error;
+
+ DEBUG(frame->worker, "<%lu> Decode HAProxy HELLO frame", client->id);
+
+ /* Retrieve flags */
+ memcpy((char *)&(frame->flags), p, 4);
+ p += 4;
+
+ /* Fragmentation is not supported for HELLO frame */
+ if (!(frame->flags & SPOE_FRM_FL_FIN)) {
+ client->status_code = SPOE_FRM_ERR_FRAG_NOT_SUPPORTED;
+ goto error;
+ }
+
+ /* stream-id and frame-id must be cleared */
+ if (*p != 0 || *(p+1) != 0) {
+ client->status_code = SPOE_FRM_ERR_INVALID;
+ goto error;
+ }
+ p += 2;
+
+ /* Loop on K/V items */
+ while (p < end) {
+ char *str;
+ uint64_t sz;
+
+ /* Decode the item name */
+ spoe_decode_buffer(&p, end, &str, &sz);
+ if (!str) {
+ client->status_code = SPOE_FRM_ERR_INVALID;
+ goto error;
+ }
+
+ /* Check "supported-versions" K/V item */
+ if (!memcmp(str, "supported-versions", sz)) {
+ if (check_proto_version(frame, &p, end) == -1) {
+ client->status_code = SPOE_FRM_ERR_INVALID;
+ goto error;
+ }
+ }
+ /* Check "max-frame-size" K/V item */
+ else if (!memcmp(str, "max-frame-size", sz)) {
+ if (check_max_frame_size(frame, &p, end) == -1) {
+ client->status_code = SPOE_FRM_ERR_INVALID;
+ goto error;
+ }
+ }
+ /* Check "healthcheck" K/V item */
+ else if (!memcmp(str, "healthcheck", sz)) {
+ if (check_healthcheck(frame, &p, end) == -1) {
+ client->status_code = SPOE_FRM_ERR_INVALID;
+ goto error;
+ }
+ }
+ /* Check "capabilities" K/V item */
+ else if (!memcmp(str, "capabilities", sz)) {
+ if (check_capabilities(frame, &p, end) == -1) {
+ client->status_code = SPOE_FRM_ERR_INVALID;
+ goto error;
+ }
+ }
+ /* Check "engine-id" K/V item */
+ else if (!memcmp(str, "engine-id", sz)) {
+ if (check_engine_id(frame, &p, end) == -1) {
+ client->status_code = SPOE_FRM_ERR_INVALID;
+ goto error;
+ }
+ }
+ else {
+ DEBUG(frame->worker, "<%lu> Skip K/V item : key=%.*s",
+ client->id, (int)sz, str);
+
+ /* Silently ignore unknown item */
+ if (spoe_skip_data(&p, end) == -1) {
+ client->status_code = SPOE_FRM_ERR_INVALID;
+ goto error;
+ }
+ }
+ }
+
+ if (async == false || client->engine_id == NULL)
+ client->async = false;
+ if (pipelining == false)
+ client->pipelining = false;
+
+ if (client->async == true)
+ use_spoe_engine(client);
+
+ return (p - frame->buf);
+ error:
+ return -1;
+}
+
+/* Decode a DISCONNECT frame received from HAProxy. It returns -1 if an error
+ * occurred, otherwise the number of read bytes. DISCONNECT frame cannot be
+ * ignored and having another frame than a DISCONNECT frame is an error.*/
+static int
+handle_hadiscon(struct spoe_frame *frame)
+{
+ struct client *client = frame->client;
+ char *p, *end;
+
+ p = frame->buf;
+ end = frame->buf + frame->len;
+
+ /* Check frame type: we really want a DISCONNECT frame */
+ if (*p++ != SPOE_FRM_T_HAPROXY_DISCON)
+ goto error;
+
+ DEBUG(frame->worker, "<%lu> Decode HAProxy DISCONNECT frame", client->id);
+
+ /* Retrieve flags */
+ memcpy((char *)&(frame->flags), p, 4);
+ p += 4;
+
+ /* Fragmentation is not supported for DISCONNECT frame */
+ if (!(frame->flags & SPOE_FRM_FL_FIN)) {
+ client->status_code = SPOE_FRM_ERR_FRAG_NOT_SUPPORTED;
+ goto error;
+ }
+
+ /* stream-id and frame-id must be cleared */
+ if (*p != 0 || *(p+1) != 0) {
+ client->status_code = SPOE_FRM_ERR_INVALID;
+ goto error;
+ }
+ p += 2;
+
+ client->status_code = SPOE_FRM_ERR_NONE;
+
+ /* Loop on K/V items */
+ while (p < end) {
+ char *str;
+ uint64_t sz;
+
+ /* Decode item key */
+ spoe_decode_buffer(&p, end, &str, &sz);
+ if (!str) {
+ client->status_code = SPOE_FRM_ERR_INVALID;
+ goto error;
+ }
+
+ /* Check "status-code" K/V item */
+ if (!memcmp(str, "status-code", sz)) {
+ if (check_discon_status_code(frame, &p, end) == -1) {
+ client->status_code = SPOE_FRM_ERR_INVALID;
+ goto error;
+ }
+ }
+ /* Check "message" K/V item */
+ else if (!memcmp(str, "message", sz)) {
+ if (check_discon_message(frame, &p, end) == -1) {
+ client->status_code = SPOE_FRM_ERR_INVALID;
+ goto error;
+ }
+ }
+ else {
+ DEBUG(frame->worker, "<%lu> Skip K/V item : key=%.*s",
+ client->id, (int)sz, str);
+
+ /* Silently ignore unknown item */
+ if (spoe_skip_data(&p, end) == -1) {
+ client->status_code = SPOE_FRM_ERR_INVALID;
+ goto error;
+ }
+ }
+ }
+
+ return (p - frame->buf);
+ error:
+ return -1;
+}
+
+/* Decode a NOTIFY frame received from HAProxy. It returns -1 if an error
+ * occurred, 0 if it must be must be ignored, otherwise the number of read
+ * bytes. */
+static int
+handle_hanotify(struct spoe_frame *frame)
+{
+ struct client *client = frame->client;
+ char *p, *end;
+ uint64_t stream_id, frame_id;
+
+ p = frame->buf;
+ end = frame->buf + frame->len;
+
+ /* Check frame type */
+ if (*p++ != SPOE_FRM_T_HAPROXY_NOTIFY)
+ goto ignore;
+
+ DEBUG(frame->worker, "<%lu> Decode HAProxy NOTIFY frame", client->id);
+
+ /* Retrieve flags */
+ memcpy((char *)&(frame->flags), p, 4);
+ p += 4;
+
+ /* Fragmentation is not supported */
+ if (!(frame->flags & SPOE_FRM_FL_FIN) && fragmentation == false) {
+ client->status_code = SPOE_FRM_ERR_FRAG_NOT_SUPPORTED;
+ goto error;
+ }
+
+ /* Read the stream-id and frame-id */
+ if (decode_varint(&p, end, &stream_id) == -1)
+ goto ignore;
+ if (decode_varint(&p, end, &frame_id) == -1)
+ goto ignore;
+
+ frame->stream_id = (unsigned int)stream_id;
+ frame->frame_id = (unsigned int)frame_id;
+
+ DEBUG(frame->worker, "<%lu> STREAM-ID=%u - FRAME-ID=%u"
+ " - %s frame received"
+ " - frag_len=%u - len=%u - offset=%ld",
+ client->id, frame->stream_id, frame->frame_id,
+ (frame->flags & SPOE_FRM_FL_FIN) ? "unfragmented" : "fragmented",
+ frame->frag_len, frame->len, p - frame->buf);
+
+ frame->fragmented = !(frame->flags & SPOE_FRM_FL_FIN);
+ frame->offset = (p - frame->buf);
+ return acc_payload(frame);
+
+ ignore:
+ return 0;
+
+ error:
+ return -1;
+}
+
+/* Decode next part of a fragmented frame received from HAProxy. It returns -1
+ * if an error occurred, 0 if it must be must be ignored, otherwise the number
+ * of read bytes. */
+static int
+handle_hafrag(struct spoe_frame *frame)
+{
+ struct client *client = frame->client;
+ char *p, *end;
+ uint64_t stream_id, frame_id;
+
+ p = frame->buf;
+ end = frame->buf + frame->len;
+
+ /* Check frame type */
+ if (*p++ != SPOE_FRM_T_UNSET)
+ goto ignore;
+
+ DEBUG(frame->worker, "<%lu> Decode Next part of a fragmented frame", client->id);
+
+ /* Fragmentation is not supported */
+ if (fragmentation == false) {
+ client->status_code = SPOE_FRM_ERR_FRAG_NOT_SUPPORTED;
+ goto error;
+ }
+
+ /* Retrieve flags */
+ memcpy((char *)&(frame->flags), p, 4);
+ p+= 4;
+
+ /* Read the stream-id and frame-id */
+ if (decode_varint(&p, end, &stream_id) == -1)
+ goto ignore;
+ if (decode_varint(&p, end, &frame_id) == -1)
+ goto ignore;
+
+ if (frame->fragmented == false ||
+ frame->stream_id != (unsigned int)stream_id ||
+ frame->frame_id != (unsigned int)frame_id) {
+ client->status_code = SPOE_FRM_ERR_INTERLACED_FRAMES;
+ goto error;
+ }
+
+ if (frame->flags & SPOE_FRM_FL_ABRT) {
+ DEBUG(frame->worker, "<%lu> STREAM-ID=%u - FRAME-ID=%u"
+ " - Abort processing of a fragmented frame"
+ " - frag_len=%u - len=%u - offset=%ld",
+ client->id, frame->stream_id, frame->frame_id,
+ frame->frag_len, frame->len, p - frame->buf);
+ goto ignore;
+ }
+
+ DEBUG(frame->worker, "<%lu> STREAM-ID=%u - FRAME-ID=%u"
+ " - %s fragment of a fragmented frame received"
+ " - frag_len=%u - len=%u - offset=%ld",
+ client->id, frame->stream_id, frame->frame_id,
+ (frame->flags & SPOE_FRM_FL_FIN) ? "last" : "next",
+ frame->frag_len, frame->len, p - frame->buf);
+
+ frame->offset = (p - frame->buf);
+ return acc_payload(frame);
+
+ ignore:
+ return 0;
+
+ error:
+ return -1;
+}
+
+/* Encode a HELLO frame to send it to HAProxy. It returns the number of written
+ * bytes. */
+static int
+prepare_agenthello(struct spoe_frame *frame)
+{
+ struct client *client = frame->client;
+ char *p, *end;
+ char capabilities[64];
+ int n;
+ unsigned int flags = SPOE_FRM_FL_FIN;
+
+ DEBUG(frame->worker, "<%lu> Encode Agent HELLO frame", client->id);
+ frame->type = SPOA_FRM_T_AGENT;
+
+ p = frame->buf;
+ end = frame->buf+max_frame_size;
+
+ /* Frame Type */
+ *p++ = SPOE_FRM_T_AGENT_HELLO;
+
+ /* Set flags */
+ memcpy(p, (char *)&flags, 4);
+ p += 4;
+
+ /* No stream-id and frame-id for HELLO frames */
+ *p++ = 0;
+ *p++ = 0;
+
+ /* "version" K/V item */
+ spoe_encode_buffer("version", 7, &p, end);
+ *p++ = SPOE_DATA_T_STR;
+ spoe_encode_buffer(SPOP_VERSION, SLEN(SPOP_VERSION), &p, end);
+ DEBUG(frame->worker, "<%lu> Agent version : %s",
+ client->id, SPOP_VERSION);
+
+
+ /* "max-frame-size" K/V item */
+ spoe_encode_buffer("max-frame-size", 14, &p ,end);
+ *p++ = SPOE_DATA_T_UINT32;
+ encode_varint(client->max_frame_size, &p, end);
+ DEBUG(frame->worker, "<%lu> Agent maximum frame size : %u",
+ client->id, client->max_frame_size);
+
+ /* "capabilities" K/V item */
+ spoe_encode_buffer("capabilities", 12, &p, end);
+ *p++ = SPOE_DATA_T_STR;
+
+ memset(capabilities, 0, sizeof(capabilities));
+ n = 0;
+
+ /* 1. Fragmentation capability ? */
+ if (fragmentation == true) {
+ memcpy(capabilities, "fragmentation", 13);
+ n += 13;
+ }
+ /* 2. Pipelining capability ? */
+ if (client->pipelining == true) {
+ if (n) capabilities[n++] = ',';
+ memcpy(capabilities + n, "pipelining", 10);
+ n += 10;
+ }
+ /* 3. Async capability ? */
+ if (client->async == true) {
+ if (n) capabilities[n++] = ',';
+ memcpy(capabilities + n, "async", 5);
+ n += 5;
+ }
+ spoe_encode_buffer(capabilities, n, &p, end);
+
+ DEBUG(frame->worker, "<%lu> Agent capabilities : %.*s",
+ client->id, n, capabilities);
+
+ frame->len = (p - frame->buf);
+ return frame->len;
+}
+
+/* Encode a DISCONNECT frame to send it to HAProxy. It returns the number of
+ * written bytes. */
+static int
+prepare_agentdicon(struct spoe_frame *frame)
+{
+ struct client *client = frame->client;
+ char *p, *end;
+ const char *reason;
+ int rlen;
+ unsigned int flags = SPOE_FRM_FL_FIN;
+
+ DEBUG(frame->worker, "<%lu> Encode Agent DISCONNECT frame", client->id);
+ frame->type = SPOA_FRM_T_AGENT;
+
+ p = frame->buf;
+ end = frame->buf+max_frame_size;
+
+ if (client->status_code >= SPOE_FRM_ERRS)
+ client->status_code = SPOE_FRM_ERR_UNKNOWN;
+ reason = spoe_frm_err_reasons[client->status_code];
+ rlen = strlen(reason);
+
+ /* Frame type */
+ *p++ = SPOE_FRM_T_AGENT_DISCON;
+
+ /* Set flags */
+ memcpy(p, (char *)&flags, 4);
+ p += 4;
+
+ /* No stream-id and frame-id for DISCONNECT frames */
+ *p++ = 0;
+ *p++ = 0;
+
+ /* There are 2 mandatory items: "status-code" and "message" */
+
+ /* "status-code" K/V item */
+ spoe_encode_buffer("status-code", 11, &p, end);
+ *p++ = SPOE_DATA_T_UINT32;
+ encode_varint(client->status_code, &p, end);
+ DEBUG(frame->worker, "<%lu> Disconnect status code : %u",
+ client->id, client->status_code);
+
+ /* "message" K/V item */
+ spoe_encode_buffer("message", 7, &p, end);
+ *p++ = SPOE_DATA_T_STR;
+ spoe_encode_buffer(reason, rlen, &p, end);
+ DEBUG(frame->worker, "<%lu> Disconnect message : %s",
+ client->id, reason);
+
+ frame->len = (p - frame->buf);
+ return frame->len;
+}
+
+/* Encode a ACK frame to send it to HAProxy. It returns the number of written
+ * bytes. */
+static int
+prepare_agentack(struct spoe_frame *frame)
+{
+ char *p, *end;
+ unsigned int flags = SPOE_FRM_FL_FIN;
+
+ /* Be careful here, in async mode, frame->client can be NULL */
+
+ DEBUG(frame->worker, "Encode Agent ACK frame");
+ frame->type = SPOA_FRM_T_AGENT;
+
+ p = frame->buf;
+ end = frame->buf+max_frame_size;
+
+ /* Frame type */
+ *p++ = SPOE_FRM_T_AGENT_ACK;
+
+ /* Set flags */
+ memcpy(p, (char *)&flags, 4);
+ p += 4;
+
+ /* Set stream-id and frame-id for ACK frames */
+ encode_varint(frame->stream_id, &p, end);
+ encode_varint(frame->frame_id, &p, end);
+
+ DEBUG(frame->worker, "STREAM-ID=%u - FRAME-ID=%u",
+ frame->stream_id, frame->frame_id);
+
+ frame->len = (p - frame->buf);
+ return frame->len;
+}
+
+static int
+create_server_socket(void)
+{
+ struct sockaddr_in listen_addr;
+ int fd, yes = 1;
+
+ fd = socket(AF_INET, SOCK_STREAM, 0);
+ if (fd < 0) {
+ LOG(&null_worker, "Failed to create service socket : %m");
+ return -1;
+ }
+
+ memset(&listen_addr, 0, sizeof(listen_addr));
+ listen_addr.sin_family = AF_INET;
+ listen_addr.sin_addr.s_addr = INADDR_ANY;
+ listen_addr.sin_port = htons(server_port);
+
+ if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(yes)) < 0 ||
+ setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &yes, sizeof(yes)) < 0) {
+ LOG(&null_worker, "Failed to set option on server socket : %m");
+ return -1;
+ }
+
+ if (bind(fd, (struct sockaddr *) &listen_addr, sizeof(listen_addr)) < 0) {
+ LOG(&null_worker, "Failed to bind server socket : %m");
+ return -1;
+ }
+
+ if (listen(fd, CONNECTION_BACKLOG) < 0) {
+ LOG(&null_worker, "Failed to listen on server socket : %m");
+ return -1;
+ }
+
+ return fd;
+}
+
+static void
+release_frame(struct spoe_frame *frame)
+{
+ struct worker *worker;
+
+ if (frame == NULL)
+ return;
+
+ if (event_pending(&frame->process_frame_event, EV_TIMEOUT, NULL))
+ event_del(&frame->process_frame_event);
+
+ worker = frame->worker;
+ LIST_DEL(&frame->list);
+ if (frame->frag_buf)
+ free(frame->frag_buf);
+ memset(frame, 0, sizeof(*frame)+max_frame_size+4);
+ LIST_ADDQ(&worker->frames, &frame->list);
+}
+
+static void
+release_client(struct client *c)
+{
+ struct spoe_frame *frame, *back;
+
+ if (c == NULL)
+ return;
+
+ DEBUG(c->worker, "<%lu> Release client", c->id);
+
+ LIST_DEL(&c->by_worker);
+ c->worker->nbclients--;
+
+ unuse_spoe_engine(c);
+ free(c->engine_id);
+
+ if (event_pending(&c->read_frame_event, EV_READ, NULL))
+ event_del(&c->read_frame_event);
+ if (event_pending(&c->write_frame_event, EV_WRITE, NULL))
+ event_del(&c->write_frame_event);
+
+ release_frame(c->incoming_frame);
+ release_frame(c->outgoing_frame);
+ list_for_each_entry_safe(frame, back, &c->processing_frames, list) {
+ release_frame(frame);
+ }
+ list_for_each_entry_safe(frame, back, &c->outgoing_frames, list) {
+ release_frame(frame);
+ }
+
+ if (c->fd >= 0)
+ close(c->fd);
+
+ free(c);
+}
+
+static void
+reset_frame(struct spoe_frame *frame)
+{
+ if (frame == NULL)
+ return;
+
+ if (frame->frag_buf)
+ free(frame->frag_buf);
+
+ frame->type = SPOA_FRM_T_UNKNOWN;
+ frame->buf = (char *)(frame->data);
+ frame->offset = 0;
+ frame->len = 0;
+ frame->stream_id = 0;
+ frame->frame_id = 0;
+ frame->flags = 0;
+ frame->hcheck = false;
+ frame->fragmented = false;
+ frame->modsec_code = -1;
+ frame->frag_buf = NULL;
+ frame->frag_len = 0;
+ LIST_INIT(&frame->list);
+}
+
+static void
+use_spoe_engine(struct client *client)
+{
+ struct spoe_engine *eng;
+
+ if (client->engine_id == NULL)
+ return;
+
+ list_for_each_entry(eng, &client->worker->engines, list) {
+ if (!strcmp(eng->id, client->engine_id))
+ goto end;
+ }
+
+ if ((eng = malloc(sizeof(*eng))) == NULL) {
+ client->async = false;
+ return;
+ }
+
+ eng->id = strdup(client->engine_id);
+ LIST_INIT(&eng->clients);
+ LIST_INIT(&eng->processing_frames);
+ LIST_INIT(&eng->outgoing_frames);
+ LIST_ADDQ(&client->worker->engines, &eng->list);
+ LOG(client->worker, "Add new SPOE engine '%s'", eng->id);
+
+ end:
+ client->engine = eng;
+ LIST_ADDQ(&eng->clients, &client->by_engine);
+}
+
+static void
+unuse_spoe_engine(struct client *client)
+{
+ struct spoe_engine *eng;
+ struct spoe_frame *frame, *back;
+
+ if (client == NULL || client->engine == NULL)
+ return;
+
+ eng = client->engine;
+ client->engine = NULL;
+ LIST_DEL(&client->by_engine);
+ if (!LIST_ISEMPTY(&eng->clients))
+ return;
+
+ LOG(client->worker, "Remove SPOE engine '%s'", eng->id);
+ LIST_DEL(&eng->list);
+
+ list_for_each_entry_safe(frame, back, &eng->processing_frames, list) {
+ release_frame(frame);
+ }
+ list_for_each_entry_safe(frame, back, &eng->outgoing_frames, list) {
+ release_frame(frame);
+ }
+ free(eng->id);
+ free(eng);
+}
+
+
+static struct spoe_frame *
+acquire_incoming_frame(struct client *client)
+{
+ struct spoe_frame *frame;
+
+ frame = client->incoming_frame;
+ if (frame != NULL)
+ return frame;
+
+ if (LIST_ISEMPTY(&client->worker->frames)) {
+ if ((frame = calloc(1, sizeof(*frame)+max_frame_size+4)) == NULL) {
+ LOG(client->worker, "Failed to allocate new frame : %m");
+ return NULL;
+ }
+ }
+ else {
+ frame = LIST_NEXT(&client->worker->frames, typeof(frame), list);
+ LIST_DEL(&frame->list);
+ }
+
+ reset_frame(frame);
+ frame->worker = client->worker;
+ frame->engine = client->engine;
+ frame->client = client;
+
+ if (event_assign(&frame->process_frame_event, client->worker->base, -1,
+ EV_TIMEOUT|EV_PERSIST, process_frame_cb, frame) < 0) {
+ LOG(client->worker, "Failed to create frame event");
+ return NULL;
+ }
+
+ client->incoming_frame = frame;
+ return frame;
+}
+
+static struct spoe_frame *
+acquire_outgoing_frame(struct client *client)
+{
+ struct spoe_engine *engine = client->engine;
+ struct spoe_frame *frame = NULL;
+
+ if (client->outgoing_frame != NULL)
+ frame = client->outgoing_frame;
+ else if (!LIST_ISEMPTY(&client->outgoing_frames)) {
+ frame = LIST_NEXT(&client->outgoing_frames, typeof(frame), list);
+ LIST_DEL(&frame->list);
+ client->outgoing_frame = frame;
+ }
+ else if (engine!= NULL && !LIST_ISEMPTY(&engine->outgoing_frames)) {
+ frame = LIST_NEXT(&engine->outgoing_frames, typeof(frame), list);
+ LIST_DEL(&frame->list);
+ client->outgoing_frame = frame;
+ }
+ return frame;
+}
+
+static void
+write_frame(struct client *client, struct spoe_frame *frame)
+{
+ uint32_t netint;
+
+ LIST_DEL(&frame->list);
+
+ frame->buf = (char *)(frame->data);
+ frame->offset = 0;
+ netint = htonl(frame->len);
+ memcpy(frame->buf, &netint, 4);
+
+ if (client != NULL) { /* HELLO or DISCONNECT frames */
+ event_add(&client->write_frame_event, NULL);
+
+ /* Try to process the frame as soon as possible, and always
+ * attach it to the client */
+ if (client->async || client->pipelining) {
+ if (client->outgoing_frame == NULL)
+ client->outgoing_frame = frame;
+ else
+ LIST_ADD(&client->outgoing_frames, &frame->list);
+ }
+ else {
+ client->outgoing_frame = frame;
+ event_del(&client->read_frame_event);
+ }
+ }
+ else { /* for all other frames */
+ if (frame->client == NULL) { /* async mode ! */
+ LIST_ADDQ(&frame->engine->outgoing_frames, &frame->list);
+ list_for_each_entry(client, &frame->engine->clients, by_engine)
+ event_add(&client->write_frame_event, NULL);
+ }
+ else if (frame->client->pipelining) {
+ LIST_ADDQ(&frame->client->outgoing_frames, &frame->list);
+ event_add(&frame->client->write_frame_event, NULL);
+ }
+ else {
+ frame->client->outgoing_frame = frame;
+ event_add(&frame->client->write_frame_event, NULL);
+ event_del(&frame->client->read_frame_event);
+ }
+ }
+}
+
+static void
+process_incoming_frame(struct spoe_frame *frame)
+{
+ struct client *client = frame->client;
+
+ if (event_add(&frame->process_frame_event, &processing_delay) < 0) {
+ LOG(client->worker, "Failed to process incoming frame");
+ release_frame(frame);
+ return;
+ }
+
+ if (client->async) {
+ frame->client = NULL;
+ LIST_ADDQ(&frame->engine->processing_frames, &frame->list);
+ }
+ else if (client->pipelining)
+ LIST_ADDQ(&client->processing_frames, &frame->list);
+ else
+ event_del(&client->read_frame_event);
+}
+
+static void
+signal_cb(evutil_socket_t sig, short events, void *user_data)
+{
+ struct event_base *base = user_data;
+ int i;
+
+ DEBUG(&null_worker, "Stopping the server");
+
+ event_base_loopbreak(base);
+ DEBUG(&null_worker, "Main event loop stopped");
+
+ for (i = 0; i < num_workers; i++) {
+ event_base_loopbreak(workers[i].base);
+ DEBUG(&null_worker, "Event loop stopped for worker %02d",
+ workers[i].id);
+ }
+}
+
+static void
+worker_monitor_cb(evutil_socket_t fd, short events, void *arg)
+{
+ struct worker *worker = arg;
+
+ LOG(worker, "%u clients connected", worker->nbclients);
+}
+
+static void
+process_frame_cb(evutil_socket_t fd, short events, void *arg)
+{
+ struct spoe_frame *frame = arg;
+ char *p, *end;
+ int ret;
+
+ DEBUG(frame->worker,
+ "Process frame messages : STREAM-ID=%u - FRAME-ID=%u - length=%u bytes",
+ frame->stream_id, frame->frame_id, frame->len - frame->offset);
+
+ p = frame->buf + frame->offset;
+ end = frame->buf + frame->len;
+
+ /* Loop on messages */
+ while (p < end) {
+ char *str;
+ uint64_t sz;
+ int nbargs;
+
+ /* Decode the message name */
+ spoe_decode_buffer(&p, end, &str, &sz);
+ if (!str)
+ goto stop_processing;
+
+ DEBUG(frame->worker, "Process SPOE Message '%.*s'", (int)sz, str);
+
+ nbargs = *p++; /* Get the number of arguments */
+ frame->offset = (p - frame->buf); /* Save index to handle errors and skip args */
+ if (!memcmp(str, "check-request", sz)) {
+ struct modsecurity_parameters params;
+
+ memset(¶ms, 0, sizeof(params));
+
+ if (nbargs != 8)
+ goto skip_message;
+
+ /* Decode parameter name. */
+ if (spoe_decode_buffer(&p, end, &str, &sz) == -1)
+ goto stop_processing;
+
+ /* Decode unique id. */
+ if (spoe_decode_data(&p, end, ¶ms.uniqueid) == -1)
+ goto skip_message;
+
+ /* Decode parameter name. */
+ if (spoe_decode_buffer(&p, end, &str, &sz) == -1)
+ goto stop_processing;
+
+ /* Decode method. */
+ if (spoe_decode_data(&p, end, ¶ms.method) == -1)
+ goto skip_message;
+
+ /* Decode parameter name. */
+ if (spoe_decode_buffer(&p, end, &str, &sz) == -1)
+ goto stop_processing;
+
+ /* Decode path. */
+ if (spoe_decode_data(&p, end, ¶ms.path) == -1)
+ goto skip_message;
+
+ /* Decode parameter name. */
+ if (spoe_decode_buffer(&p, end, &str, &sz) == -1)
+ goto stop_processing;
+
+ /* Decode query. */
+ if (spoe_decode_data(&p, end, ¶ms.query) == -1)
+ goto skip_message;
+
+ /* Decode parameter name. */
+ if (spoe_decode_buffer(&p, end, &str, &sz) == -1)
+ goto stop_processing;
+
+ /* Decode protocol version. */
+ if (spoe_decode_data(&p, end, ¶ms.vers) == -1)
+ goto skip_message;
+
+ /* Decode parameter name. */
+ if (spoe_decode_buffer(&p, end, &str, &sz) == -1)
+ goto stop_processing;
+
+ /* Decode hdrs. */
+ if (spoe_decode_data(&p, end, ¶ms.hdrs_bin) == -1)
+ goto skip_message;
+
+ /* Decode parameter name. */
+ if (spoe_decode_buffer(&p, end, &str, &sz) == -1)
+ goto stop_processing;
+
+ /* Decode body length. */
+ if (spoe_decode_data(&p, end, ¶ms.body_length) == -1)
+ goto skip_message;
+
+ /* Decode parameter name. */
+ if (spoe_decode_buffer(&p, end, &str, &sz) == -1)
+ goto stop_processing;
+
+ /* Decode body. */
+ if (spoe_decode_data(&p, end, ¶ms.body) == -1)
+ goto skip_message;
+
+ frame->modsec_code = modsecurity_process(frame->worker, ¶ms);
+ }
+ else {
+ skip_message:
+ p = frame->buf + frame->offset; /* Restore index */
+
+ while (nbargs-- > 0) {
+ /* Silently ignore argument: its name and its value */
+ if (spoe_decode_buffer(&p, end, &str, &sz) == -1)
+ goto stop_processing;
+ if (spoe_skip_data(&p, end) == -1)
+ goto stop_processing;
+ }
+ }
+ }
+
+ stop_processing:
+ /* Prepare agent ACK frame */
+ frame->buf = (char *)(frame->data) + 4;
+ frame->offset = 0;
+ frame->len = 0;
+ frame->flags = 0;
+
+ ret = prepare_agentack(frame);
+ p = frame->buf + ret;
+
+ if (frame->modsec_code != -1) {
+ DEBUG(frame->worker, "Add action : set variable code=%u",
+ frame->modsec_code);
+
+ *p++ = SPOE_ACT_T_SET_VAR; /* Action type */
+ *p++ = 3; /* Number of args */
+ *p++ = SPOE_SCOPE_TXN; /* Arg 1: the scope */
+ spoe_encode_buffer("code", 8, &p, end); /* Arg 2: variable name */
+ *p++ = SPOE_DATA_T_UINT32;
+ encode_varint(frame->modsec_code, &p, end); /* Arg 3: variable value */
+ frame->len = (p - frame->buf);
+ }
+ write_frame(NULL, frame);
+}
+
+static void
+read_frame_cb(evutil_socket_t fd, short events, void *arg)
+{
+ struct client *client = arg;
+ struct spoe_frame *frame;
+ uint32_t netint;
+ int n;
+
+ DEBUG(client->worker, "<%lu> %s", client->id, __FUNCTION__);
+ if ((frame = acquire_incoming_frame(client)) == NULL)
+ goto close;
+
+ frame->type = SPOA_FRM_T_HAPROXY;
+ if (frame->buf == (char *)(frame->data)) {
+ /* Read the frame length: frame->buf points on length part (frame->data) */
+ n = read(client->fd, frame->buf+frame->offset, 4-frame->offset);
+ if (n <= 0) {
+ if (n < 0)
+ LOG(client->worker, "Failed to read frame length : %m");
+ goto close;
+ }
+ frame->offset += n;
+ if (frame->offset != 4)
+ return;
+ memcpy(&netint, frame->buf, 4);
+ frame->buf += 4;
+ frame->offset = 0;
+ frame->len = ntohl(netint);
+ }
+
+ /* Read the frame: frame->buf points on frame part (frame->data+4)*/
+ n = read(client->fd, frame->buf + frame->offset,
+ frame->len - frame->offset);
+ if (n <= 0) {
+ if (n < 0) {
+ LOG(client->worker, "Frame to read frame : %m");
+ goto close;
+ }
+ return;
+ }
+ frame->offset += n;
+ if (frame->offset != frame->len)
+ return;
+ frame->offset = 0;
+
+ DEBUG(client->worker, "<%lu> New Frame of %u bytes received",
+ client->id, frame->len);
+
+ switch (client->state) {
+ case SPOA_ST_CONNECTING:
+ if (handle_hahello(frame) < 0) {
+ LOG(client->worker, "Failed to decode HELLO frame");
+ goto disconnect;
+ }
+ prepare_agenthello(frame);
+ goto write_frame;
+
+ case SPOA_ST_PROCESSING:
+ if (frame->buf[0] == SPOE_FRM_T_HAPROXY_DISCON) {
+ client->state = SPOA_ST_DISCONNECTING;
+ goto disconnecting;
+ }
+ if (frame->buf[0] == SPOE_FRM_T_UNSET)
+ n = handle_hafrag(frame);
+ else
+ n = handle_hanotify(frame);
+
+ if (n < 0) {
+ LOG(client->worker, "Failed to decode frame: %s",
+ spoe_frm_err_reasons[client->status_code]);
+ goto disconnect;
+ }
+ else if (n == 0) {
+ LOG(client->worker, "Ignore invalid/unknown/aborted frame");
+ goto ignore_frame;
+ }
+ else if (n == 1)
+ goto noop;
+ else
+ goto process_frame;
+
+ case SPOA_ST_DISCONNECTING:
+ disconnecting:
+ if (handle_hadiscon(frame) < 0) {
+ LOG(client->worker, "Failed to decode DISCONNECT frame");
+ goto disconnect;
+ }
+ if (client->status_code != SPOE_FRM_ERR_NONE)
+ LOG(client->worker, "<%lu> Peer closed connection: %s",
+ client->id, spoe_frm_err_reasons[client->status_code]);
+ client->status_code = SPOE_FRM_ERR_NONE;
+ goto disconnect;
+ }
+
+ noop:
+ return;
+
+ ignore_frame:
+ reset_frame(frame);
+ return;
+
+ process_frame:
+ process_incoming_frame(frame);
+ client->incoming_frame = NULL;
+ return;
+
+ write_frame:
+ write_frame(client, frame);
+ client->incoming_frame = NULL;
+ return;
+
+ disconnect:
+ client->state = SPOA_ST_DISCONNECTING;
+ if (prepare_agentdicon(frame) < 0) {
+ LOG(client->worker, "Failed to encode DISCONNECT frame");
+ goto close;
+ }
+ goto write_frame;
+
+ close:
+ release_client(client);
+}
+
+static void
+write_frame_cb(evutil_socket_t fd, short events, void *arg)
+{
+ struct client *client = arg;
+ struct spoe_frame *frame;
+ int n;
+
+ DEBUG(client->worker, "<%lu> %s", client->id, __FUNCTION__);
+ if ((frame = acquire_outgoing_frame(client)) == NULL) {
+ event_del(&client->write_frame_event);
+ return;
+ }
+
+ if (frame->buf == (char *)(frame->data)) {
+ /* Write the frame length: frame->buf points on length part (frame->data) */
+ n = write(client->fd, frame->buf+frame->offset, 4-frame->offset);
+ if (n <= 0) {
+ if (n < 0)
+ LOG(client->worker, "Failed to write frame length : %m");
+ goto close;
+ }
+ frame->offset += n;
+ if (frame->offset != 4)
+ return;
+ frame->buf += 4;
+ frame->offset = 0;
+ }
+
+ /* Write the frame: frame->buf points on frame part (frame->data+4)*/
+ n = write(client->fd, frame->buf + frame->offset,
+ frame->len - frame->offset);
+ if (n <= 0) {
+ if (n < 0) {
+ LOG(client->worker, "Failed to write frame : %m");
+ goto close;
+ }
+ return;
+ }
+ frame->offset += n;
+ if (frame->offset != frame->len)
+ return;
+
+ DEBUG(client->worker, "<%lu> Frame of %u bytes send",
+ client->id, frame->len);
+
+ switch (client->state) {
+ case SPOA_ST_CONNECTING:
+ if (frame->hcheck == true) {
+ DEBUG(client->worker,
+ "<%lu> Close client after healthcheck",
+ client->id);
+ goto close;
+ }
+ client->state = SPOA_ST_PROCESSING;
+ break;
+
+ case SPOA_ST_PROCESSING:
+ break;
+
+ case SPOA_ST_DISCONNECTING:
+ goto close;
+ }
+
+ release_frame(frame);
+ client->outgoing_frame = NULL;
+ if (!client->async && !client->pipelining) {
+ event_del(&client->write_frame_event);
+ event_add(&client->read_frame_event, NULL);
+ }
+ return;
+
+ close:
+ release_client(client);
+}
+
+static void
+accept_cb(int listener, short event, void *arg)
+{
+ struct worker *worker;
+ struct client *client;
+ int fd;
+
+ worker = &workers[clicount++ % num_workers];
+
+ if ((fd = accept(listener, NULL, NULL)) < 0) {
+ if (errno != EAGAIN && errno != EWOULDBLOCK)
+ LOG(worker, "Failed to accept client connection : %m");
+ return;
+ }
+
+ DEBUG(&null_worker,
+ "<%lu> New Client connection accepted and assigned to worker %02d",
+ clicount, worker->id);
+
+ if (evutil_make_socket_nonblocking(fd) < 0) {
+ LOG(&null_worker, "Failed to set client socket to non-blocking : %m");
+ close(fd);
+ return;
+ }
+
+ if ((client = calloc(1, sizeof(*client))) == NULL) {
+ LOG(&null_worker, "Failed to allocate memory for client state : %m");
+ close(fd);
+ return;
+ }
+
+ client->id = clicount;
+ client->fd = fd;
+ client->worker = worker;
+ client->state = SPOA_ST_CONNECTING;
+ client->status_code = SPOE_FRM_ERR_NONE;
+ client->max_frame_size = max_frame_size;
+ client->engine = NULL;
+ client->pipelining = false;
+ client->async = false;
+ client->incoming_frame = NULL;
+ client->outgoing_frame = NULL;
+ LIST_INIT(&client->processing_frames);
+ LIST_INIT(&client->outgoing_frames);
+
+ LIST_ADDQ(&worker->clients, &client->by_worker);
+
+ worker->nbclients++;
+
+ if (event_assign(&client->read_frame_event, worker->base, fd,
+ EV_READ|EV_PERSIST, read_frame_cb, client) < 0 ||
+ event_assign(&client->write_frame_event, worker->base, fd,
+ EV_WRITE|EV_PERSIST, write_frame_cb, client) < 0) {
+ LOG(&null_worker, "Failed to create client events");
+ release_client(client);
+ return;
+ }
+ event_add(&client->read_frame_event, NULL);
+}
+
+static void *
+worker_function(void *data)
+{
+ struct client *client, *cback;
+ struct spoe_frame *frame, *fback;
+ struct worker *worker = data;
+
+ DEBUG(worker, "Worker ready to process client messages");
+ event_base_dispatch(worker->base);
+
+ list_for_each_entry_safe(client, cback, &worker->clients, by_worker) {
+ release_client(client);
+ }
+
+ list_for_each_entry_safe(frame, fback, &worker->frames, list) {
+ LIST_DEL(&frame->list);
+ free(frame);
+ }
+
+ event_free(worker->monitor_event);
+ event_base_free(worker->base);
+ DEBUG(worker, "Worker is stopped");
+ pthread_exit(&null_worker);
+}
+
+
+static int
+parse_processing_delay(const char *str)
+{
+ unsigned long value;
+
+ value = 0;
+ while (1) {
+ unsigned int j;
+
+ j = *str - '0';
+ if (j > 9)
+ break;
+ str++;
+ value *= 10;
+ value += j;
+ }
+
+ switch (*str) {
+ case '\0': /* no unit = millisecond */
+ value *= 1000;
+ break;
+ case 's': /* second */
+ value *= 1000000;
+ str++;
+ break;
+ case 'm': /* millisecond : "ms" */
+ if (str[1] != 's')
+ return -1;
+ value *= 1000;
+ str += 2;
+ break;
+ case 'u': /* microsecond : "us" */
+ if (str[1] != 's')
+ return -1;
+ str += 2;
+ break;
+ default:
+ return -1;
+ }
+ if (*str)
+ return -1;
+
+ processing_delay.tv_sec = (time_t)(value / 1000000);
+ processing_delay.tv_usec = (suseconds_t)(value % 1000000);
+ return 0;
+}
+
+
+static void
+usage(char *prog)
+{
+ fprintf(stderr,
+ "Usage : %s [OPTION]...\n"
+ " -h Print this message\n"
+ " -d Enable the debug mode\n"
+ " -f <config-file> ModSecurity configuration file\n"
+ " -m <max-frame-size> Specify the maximum frame size (default : %u)\n"
+ " -p <port> Specify the port to listen on (default : %d)\n"
+ " -n <num-workers> Specify the number of workers (default : %d)\n"
+ " -c <capability> Enable the support of the specified capability\n"
+ " -t <time> Set a delay to process a message (default: 0)\n"
+ " The value is specified in milliseconds by default,\n"
+ " but can be in any other unit if the number is suffixed\n"
+ " by a unit (us, ms, s)\n"
+ "\n"
+ " Supported capabilities: fragmentation, pipelining, async\n",
+ prog, MAX_FRAME_SIZE, DEFAULT_PORT, NUM_WORKERS);
+}
+
+int
+main(int argc, char **argv)
+{
+ struct event_base *base = NULL;
+ struct event *signal_event = NULL, *accept_event = NULL;
+ int opt, i, fd = -1;
+ const char *configuration_file = NULL;
+
+ // TODO: add '-t <processing-time>' option
+ while ((opt = getopt(argc, argv, "hdm:n:p:c:t:f:")) != -1) {
+ switch (opt) {
+ case 'h':
+ usage(argv[0]);
+ return EXIT_SUCCESS;
+ case 'd':
+ debug = true;
+ break;
+ case 'm':
+ max_frame_size = atoi(optarg);
+ break;
+ case 'n':
+ num_workers = atoi(optarg);
+ break;
+ case 'p':
+ server_port = atoi(optarg);
+ break;
+ case 'f':
+ configuration_file = optarg;
+ break;
+ case 'c':
+ if (!strcmp(optarg, "pipelining"))
+ pipelining = true;
+ else if (!strcmp(optarg, "async"))
+ async = true;
+ else if (!strcmp(optarg, "fragmentation"))
+ fragmentation = true;
+ else
+ fprintf(stderr, "WARNING: unsupported capability '%s'\n", optarg);
+ break;
+ case 't':
+ if (!parse_processing_delay(optarg))
+ break;
+ fprintf(stderr, "%s: failed to parse time '%s'.\n", argv[0], optarg);
+ fprintf(stderr, "Try '%s -h' for more information.\n", argv[0]);
+ return EXIT_FAILURE;
+ default:
+ usage(argv[0]);
+ return EXIT_FAILURE;
+ }
+ }
+
+ if (!configuration_file) {
+ LOG(&null_worker, "ModSecurity configuration is required.\n");
+ goto error;
+ }
+
+ if (modsecurity_load(configuration_file) == -1)
+ goto error;
+
+ if (num_workers <= 0) {
+ LOG(&null_worker, "%s : Invalid number of workers '%d'\n",
+ argv[0], num_workers);
+ goto error;
+ }
+
+ if (server_port <= 0) {
+ LOG(&null_worker, "%s : Invalid port '%d'\n",
+ argv[0], server_port);
+ goto error;
+ }
+
+
+ if (evthread_use_pthreads() < 0) {
+ LOG(&null_worker, "No pthreads support for libevent");
+ goto error;
+ }
+
+ if ((base = event_base_new()) == NULL) {
+ LOG(&null_worker, "Failed to initialize libevent : %m");
+ goto error;
+ }
+
+ signal(SIGPIPE, SIG_IGN);
+
+ if ((fd = create_server_socket()) < 0) {
+ LOG(&null_worker, "Failed to create server socket");
+ goto error;
+ }
+ if (evutil_make_socket_nonblocking(fd) < 0) {
+ LOG(&null_worker, "Failed to set server socket to non-blocking");
+ goto error;
+ }
+
+ if ((workers = calloc(num_workers, sizeof(*workers))) == NULL) {
+ LOG(&null_worker, "Failed to set allocate memory for workers");
+ goto error;
+ }
+
+ for (i = 0; i < num_workers; ++i) {
+ struct worker *w = &workers[i];
+
+ w->id = i+1;
+ w->nbclients = 0;
+ LIST_INIT(&w->engines);
+ LIST_INIT(&w->clients);
+ LIST_INIT(&w->frames);
+
+ if ((w->base = event_base_new()) == NULL) {
+ LOG(&null_worker,
+ "Failed to initialize libevent for worker %02d : %m",
+ w->id);
+ goto error;
+ }
+
+ w->monitor_event = event_new(w->base, fd, EV_PERSIST,
+ worker_monitor_cb, (void *)w);
+ if (w->monitor_event == NULL ||
+ event_add(w->monitor_event, (struct timeval[]){{5,0}}) < 0) {
+ LOG(&null_worker,
+ "Failed to create monitor event for worker %02d",
+ w->id);
+ goto error;
+ }
+
+ if (pthread_create(&w->thread, NULL, worker_function, (void *)w)) {
+ LOG(&null_worker,
+ "Failed to start thread for worker %02d : %m",
+ w->id);
+ }
+ DEBUG(&null_worker, "Worker %02d initialized", w->id);
+ }
+
+ accept_event = event_new(base, fd, EV_READ|EV_PERSIST, accept_cb,
+ (void *)base);
+ if (accept_event == NULL || event_add(accept_event, NULL) < 0) {
+ LOG(&null_worker, "Failed to create accept event : %m");
+ }
+
+ signal_event = evsignal_new(base, SIGINT, signal_cb, (void *)base);
+ if (signal_event == NULL || event_add(signal_event, NULL) < 0) {
+ LOG(&null_worker, "Failed to create signal event : %m");
+ }
+
+ DEBUG(&null_worker,
+ "Server is ready"
+ " [fragmentation=%s - pipelining=%s - async=%s - debug=%s - max-frame-size=%u]",
+ (fragmentation?"true":"false"), (pipelining?"true":"false"), (async?"true":"false"),
+ (debug?"true":"false"), max_frame_size);
+ event_base_dispatch(base);
+
+ for (i = 0; i < num_workers; i++) {
+ struct worker *w = &workers[i];
+
+ pthread_join(w->thread, NULL);
+ DEBUG(&null_worker, "Worker %02d terminated", w->id);
+ }
+
+ free(workers);
+ event_free(signal_event);
+ event_free(accept_event);
+ event_base_free(base);
+ close(fd);
+ return EXIT_SUCCESS;
+
+ error:
+ if (workers != NULL)
+ free(workers);
+ if (signal_event != NULL)
+ event_free(signal_event);
+ if (accept_event != NULL)
+ event_free(accept_event);
+ if (base != NULL)
+ event_base_free(base);
+ if (fd != -1)
+ close(fd);
+ return EXIT_FAILURE;
+}
diff --git a/contrib/modsecurity/spoa.h b/contrib/modsecurity/spoa.h
new file mode 100644
index 0000000..daee6d2
--- /dev/null
+++ b/contrib/modsecurity/spoa.h
@@ -0,0 +1,53 @@
+/*
+ * Modsecurity wrapper for haproxy
+ *
+ * This file contains the headers of the bootstrap for laucnching and scheduling
+ * modsecurity for working with HAProxy SPOE protocol.
+ *
+ * Copyright 2016 OZON, Thierry Fournier <thierry.fourn...@ozon.io>
+ *
+ * This file is inherited from "A Random IP reputation service acting as a Stream
+ * Processing Offload Agent"
+ *
+ * Copyright 2016 HAProxy Technologies, Christopher Faulet <cfau...@haproxy.com>
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License
+ * as published by the Free Software Foundation; either version
+ * 2 of the License, or (at your option) any later version.
+ *
+ */
+#ifndef __SPOA_H__
+#define __SPOA_H__
+
+#include <event2/util.h>
+#include <event2/event.h>
+#include <event2/event_struct.h>
+#include <event2/thread.h>
+
+struct worker {
+ pthread_t thread;
+ int id;
+ struct event_base *base;
+ struct event *monitor_event;
+
+ struct list engines;
+
+ unsigned int nbclients;
+ struct list clients;
+
+ struct list frames;
+};
+
+#define LOG(worker, fmt, args...) \
+ do { \
+ struct timeval now; \
+ \
+ gettimeofday(&now, NULL); \
+ fprintf(stderr, "%ld.%06ld [%02d] " fmt "\n", \
+ now.tv_sec, now.tv_usec, (worker)->id, ##args); \
+ } while (0)
+
+#endif /* __SPOA_H__ */
+
+extern struct worker null_worker;
--
1.7.10.4
