Hi list
I join one usage of HAProxy / SPOE, it is WAF offloading.
These patches are a first version, it have some limitations describe
in the README file in the directory contrib/modsecurity.
- Christopher, please check the patch "BUG/MINOR", it is about spoe
functions.
- The exemple of ModSecurity compilation can be improved. It is based
on my local distro.
The feedback are welcome.
Thierry
>From 55702d5b7b3aa72f1e2befaa3edb5b5ccbb302f5 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/3] 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 41bc8016bb0a4ed6ce5970d8b4ed6055737b9dad 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 2/3] MINOR: Add binary encoding request sample fetch
This sample fetch encodes the full http request available in the buffer.
The request may contain or not the body, and the body may be partial. This
sample-fetch is useful with SPOE.
---
doc/configuration.txt | 15 ++++
src/proto_http.c | 240 +++++++++++++++++++++++++++++++++++++++++++++++++
2 files changed, 255 insertions(+)
diff --git a/doc/configuration.txt b/doc/configuration.txt
index 81b641e..3229ee1 100644
--- a/doc/configuration.txt
+++ b/doc/configuration.txt
@@ -14116,6 +14116,21 @@ 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.bin_enc : binary
+ Returns the current request contained in the buffer in preparsed binary form.
+ This is usefull for offloading some processing with SPOE. The format is easy
+ to understand. All the elements listed below are always present in the
+ encoding, if these elements are not in the request their length is 0. Each
+ string is described by a length followed by a number of bytes declared in the
+ length.
+
+ <str:method><str:path><str:query-string><str:protocol><int:number-of-headers>
+ *(<str:header-name><str:header-value>)<bool:full-body><str:body>
+
+ int: refers to the SPOE doc for understanding the encoding integers.
+ str: <int:length><bytes>
+ bool: 1 byte set to 0 or 1.
+
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..f93b70d 100644
--- a/src/proto_http.c
+++ b/src/proto_http.c
@@ -66,6 +66,7 @@
#include <proto/queue.h>
#include <proto/sample.h>
#include <proto/server.h>
+#include <proto/spoe.h>
#include <proto/stream.h>
#include <proto/stream_interface.h>
#include <proto/task.h>
@@ -10435,6 +10436,243 @@ smp_fetch_uniqueid(const struct arg *args, struct sample *smp, const char *kw, v
return 1;
}
+/* Returns the fulle request and the body contained in the buffer 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 fourth first couple length / string are the method, the path, the
+ * query-string and the protocol version.
+ *
+ * Next, 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"
+ *
+ * Next part is byte set to 1 if the body is full or 0 if it is partial.
+ *
+ * The last part is a length value followed by the body bytes.
+ */
+static int
+smp_fetch_bin_enc(const struct arg *args, struct sample *smp, const char *kw, void *private)
+{
+ struct http_msg *msg;
+ unsigned long body_len;
+ unsigned long block1;
+ char *body;
+ 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;
+ int header_count;
+ char *header_length_position;
+ struct http_txn *txn;
+ const char *path;
+ int path_len;
+ const char *qs;
+ int qs_len;
+ const char *vers;
+ int vers_len;
+ int i;
+ char *buf;
+ char *end;
+ int remain;
+
+ 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 method. */
+ ret = spoe_encode_varint(txn->req.sl.rq.m_l, &buf, end);
+ if (ret == -1)
+ return 0;
+ if (buf + txn->req.sl.rq.m_l > end)
+ return 0;
+ memcpy(buf, txn->req.chn->buf->p, txn->req.sl.rq.m_l);
+ buf += txn->req.sl.rq.m_l;
+
+ /* Split path and args. */
+ p = txn->req.chn->buf->p + txn->req.sl.rq.u;
+ path = p;
+ path_len = txn->req.sl.rq.u_l;
+ qs = NULL;
+ qs_len = 0;
+ for (i = 0; i < txn->req.sl.rq.u_l; i++) {
+ if (p[i] == '?') {
+ path_len = i;
+ qs = &p[i + 1];
+ qs_len = txn->req.sl.rq.u_l - i - 1;
+ break;
+ }
+ }
+
+ /* Encode the path. */
+ ret = spoe_encode_varint(path_len, &buf, end);
+ if (ret == -1)
+ return 0;
+ if (buf + path_len > end)
+ return 0;
+ memcpy(buf, path, path_len);
+ buf += path_len;
+
+ /* Encode the args. */
+ ret = spoe_encode_varint(qs_len, &buf, end);
+ if (ret == -1)
+ return 0;
+ if (buf + qs_len > end)
+ return 0;
+ memcpy(buf, qs, qs_len);
+ buf += qs_len;
+
+ /* Encode the version. */
+ vers = txn->req.chn->buf->p + txn->req.sl.rq.v;
+ vers_len = txn->req.sl.rq.v_l;
+ ret = spoe_encode_varint(vers_len, &buf, end);
+ if (ret == -1)
+ return 0;
+ if (buf + vers_len > end)
+ return 0;
+ memcpy(buf, vers, vers_len);
+ buf += vers_len;
+
+ /* We assume a max of 239 headers. This ensure the header number
+ * encoding on only one byte. If more than 239 headers are encoded,
+ * the sample-fetch fails.
+ */
+ if (buf + 1 > end)
+ return 0;
+ header_length_position = buf;
+ buf++;
+
+ /* Build array of headers. */
+ old_idx = 0;
+ cur_next = msg->chn->buf->p + hdr_idx_first_pos(idx);
+ header_count = 0;
+ 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 = spoe_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 = spoe_encode_varint(hvl, &buf, end);
+ if (ret == -1)
+ return 0;
+ if (buf + hvl > end)
+ return 0;
+ memcpy(buf, hv, hvl);
+ buf += hvl;
+
+ /* Count headers in order to repport the value at the start
+ * of the encoding.
+ */
+ header_count++;
+ }
+
+ /* Set the number of headers. */
+ if (header_count > 239)
+ return 0;
+ *header_length_position = header_count;
+
+ /* Set the body full flag. Assumes that at max 6 bytes
+ * are used for encoding length and the full flag.
+ */
+ if (buf + 1 > end)
+ return 0;
+ body_len = http_body_bytes(msg);
+ remain = msg->chn->buf->size;
+ remain -= msg->chn->buf->o; /* Output size */
+ remain -= msg->chn->buf->i; /* Input size */
+ remain -= global.tune.maxrewrite; /* Rewrite reserved size */
+ *buf = remain > 0 && buf + body_len + 6 <= end;
+ buf++;
+
+ /* Set the body length. */
+ if (buf + body_len + 5 > end)
+ body_len = end - buf - 5;
+ if (body_len < 0)
+ return 0;
+ ret = spoe_encode_varint(body_len, &buf, end);
+ if (ret == -1)
+ return 0;
+ if (buf + body_len > end)
+ return 0;
+
+ /* Copy the body. */
+ body = b_ptr(msg->chn->buf, -http_data_rewind(msg));
+
+ block1 = body_len;
+ if (block1 > msg->chn->buf->data + msg->chn->buf->size - body)
+ block1 = msg->chn->buf->data + msg->chn->buf->size - body;
+
+ if (block1 == body_len) {
+ /* buffer is not wrapped (or empty) */
+ memcpy(buf, body, body_len);
+ }
+ else {
+ /* buffer is wrapped, we need to defragment it */
+ memcpy(buf, body, block1);
+ memcpy(buf + block1, msg->chn->buf->data, body_len - block1);
+ }
+ buf += body_len;
+
+ /* 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 +13583,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.bin_enc", smp_fetch_bin_enc, 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 a6556ae766144d2ce3ccbf8516a5ba8f5e936810 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 3/3] 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 | 643 ++++++++++++
contrib/modsecurity/modsec_wrapper.h | 21 +
contrib/modsecurity/spoa.c | 1866 ++++++++++++++++++++++++++++++++++
contrib/modsecurity/spoa.h | 53 +
6 files changed, 2760 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..8aacf6c
--- /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 req.full_bin unique-id
+ 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..87ab0fc
--- /dev/null
+++ b/contrib/modsecurity/modsec_wrapper.c
@@ -0,0 +1,643 @@
+/*
+ * 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 "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 sample *request,
+ struct sample *uniqueid)
+{
+ 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 *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 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;
+ int body_full;
+
+ buf = request->data.u.str.str;
+ end = buf + request->data.u.str.len;
+
+ /* Decode method. */
+ ret = spoe_decode_varint(&buf, end, &meth_len);
+ if (ret == -1)
+ return -1;
+ meth = buf;
+ buf += meth_len;
+ if (buf > end)
+ return -1;
+
+ /* Decode path. */
+ ret = spoe_decode_varint(&buf, end, &path_len);
+ if (ret == -1)
+ return -1;
+ path = buf;
+ buf += path_len;
+ if (buf > end)
+ return -1;
+
+ /* Decode query string. */
+ ret = spoe_decode_varint(&buf, end, &qs_len);
+ if (ret == -1)
+ return -1;
+ qs = buf;
+ buf += qs_len;
+ if (buf > end)
+ return -1;
+
+ /* Decode version. */
+ ret = spoe_decode_varint(&buf, end, &vers_len);
+ if (ret == -1)
+ return -1;
+ vers = buf;
+ buf += vers_len;
+ if (buf > end)
+ return -1;
+
+ /* Decode number of headers. */
+ ret = spoe_decode_varint(&buf, end, &hdr_nb);
+ if (ret == -1)
+ return -1;
+ if (hdr_nb > 239)
+ return -1;
+
+ /* Decode each header. */
+ for (i = 0; i < hdr_nb; i++) {
+
+ /* Decode header name. */
+ ret = spoe_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 = spoe_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;
+ }
+
+ /* Get the falg body_full. */
+ body_full = *buf;
+ buf++;
+ if (buf > end)
+ return -1;
+
+ /* Decode body. */
+ ret = spoe_decode_varint(&buf, end, &body_len);
+ if (ret == -1)
+ return -1;
+ body = buf;
+ buf += body_len;
+ if (buf > end)
+ return -1;
+
+ 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->data.u.str.str, uniqueid->data.u.str.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..22839ba
--- /dev/null
+++ b/contrib/modsecurity/modsec_wrapper.h
@@ -0,0 +1,21 @@
+/*
+ * 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__
+
+int modsecurity_load(const char *file);
+int modsecurity_process(struct worker *worker, struct sample *smp, struct sample *uniqueid);
+
+#endif /* __MODSEC_WRAPPER_H__ */
diff --git a/contrib/modsecurity/spoa.c b/contrib/modsecurity/spoa.c
new file mode 100644
index 0000000..bc54857
--- /dev/null
+++ b/contrib/modsecurity/spoa.c
@@ -0,0 +1,1866 @@
+/*
+ * 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 (spoe_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 (spoe_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 (spoe_decode_varint(&p, end, &stream_id) == -1)
+ goto ignore;
+ if (spoe_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 (spoe_decode_varint(&p, end, &stream_id) == -1)
+ goto ignore;
+ if (spoe_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;
+ spoe_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;
+ spoe_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 */
+ spoe_encode_varint(frame->stream_id, &p, end);
+ spoe_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 sample req;
+ struct sample uniqueid;
+
+ memset(&req, 0, sizeof(req));
+ memset(&uniqueid, 0, sizeof(uniqueid));
+
+ if (nbargs != 2)
+ goto skip_message;
+
+ if (spoe_decode_buffer(&p, end, &str, &sz) == -1)
+ goto stop_processing;
+
+ if (spoe_decode_data(&p, end, &req) == -1)
+ goto skip_message;
+
+ if (spoe_decode_buffer(&p, end, &str, &sz) == -1)
+ goto stop_processing;
+
+ if (spoe_decode_data(&p, end, &uniqueid) == -1)
+ goto skip_message;
+
+ frame->modsec_code = modsecurity_process(frame->worker, &req, &uniqueid);
+ }
+ 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;
+ spoe_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
