MAJOR: filters: Add filters support
This patch adds the support of filters in HAProxy. The main idea is to have a
way to "easely" extend HAProxy by adding some "modules", called filters, that
will be able to change HAProxy behavior in a programmatic way.
To do so, many entry points has been added in code to let filters to hook up to
different steps of the processing. A filter must define a flt_ops sutrctures
(see include/types/filters.h for details). This structure contains all available
callbacks that a filter can define:
struct flt_ops {
/*
* Callbacks to manage the filter lifecycle
*/
int (*init) (struct proxy *p);
void (*deinit)(struct proxy *p);
int (*check) (struct proxy *p);
/*
* Stream callbacks
*/
void (*stream_start) (struct stream *s);
void (*stream_accept) (struct stream *s);
void (*session_establish)(struct stream *s);
void (*stream_stop) (struct stream *s);
/*
* HTTP callbacks
*/
int (*http_start) (struct stream *s, struct http_msg *msg);
int (*http_start_body) (struct stream *s, struct http_msg *msg);
int (*http_start_chunk) (struct stream *s, struct http_msg *msg);
int (*http_data) (struct stream *s, struct http_msg *msg);
int (*http_last_chunk) (struct stream *s, struct http_msg *msg);
int (*http_end_chunk) (struct stream *s, struct http_msg *msg);
int (*http_chunk_trailers)(struct stream *s, struct http_msg *msg);
int (*http_end_body) (struct stream *s, struct http_msg *msg);
void (*http_end) (struct stream *s, struct http_msg *msg);
void (*http_reset) (struct stream *s, struct http_msg *msg);
int (*http_pre_process) (struct stream *s, struct http_msg *msg);
int (*http_post_process) (struct stream *s, struct http_msg *msg);
void (*http_reply) (struct stream *s, short status,
const struct chunk *msg);
};
To declare and use a filter, in the configuration, the "filter" keyword must be
used in a listener/frontend section:
frontend test
...
filter <FILTER-NAME> [OPTIONS...]
The filter referenced by the <FILTER-NAME> must declare a configuration parser
on its own name to fill flt_ops and filter_conf field in the proxy's
structure. An exemple will be provided later to make it perfectly clear.
For now, filters cannot be used in backend section. But this is only a matter of
time. Documentation will also be added later. This is the first commit of a long
list about filters.
It is possible to have several filters on the same listener/frontend. These
filters are stored in an array of at most MAX_FILTERS elements (define in
include/types/filters.h). Again, this will be replaced later by a list of
filters.
The filter API has been highly refactored. Main changes are:
* Now, HA supports an infinite number of filters per proxy. To do so, filters
are stored in list.
* Because filters are stored in list, filters state has been moved from the
channel structure to the filter structure. This is cleaner because there is no
more info about filters in channel structure.
* It is possible to defined filters on backends only. For such filters,
stream_start/stream_stop callbacks are not called. Of course, it is possible
to mix frontend and backend filters.
* Now, TCP streams are also filtered. All callbacks without the 'http_' prefix
are called for all kind of streams. In addition, 2 new callbacks were added to
filter data exchanged through a TCP stream:
- tcp_data: it is called when new data are available or when old unprocessed
data are still waiting.
- tcp_forward_data: it is called when some data can be consumed.
* New callbacks attached to channel were added:
- channel_start_analyze: it is called when a filter is ready to process data
exchanged through a channel. 2 new analyzers (a frontend and a backend)
are attached to channels to call this callback. For a frontend filter, it
is called before any other analyzer. For a backend filter, it is called
when a backend is attached to a stream. So some processing cannot be
filtered in that case.
- channel_analyze: it is called before each analyzer attached to a channel,
expects analyzers responsible for data sending.
- channel_end_analyze: it is called when all other analyzers have finished
their processing. A new analyzers is attached to channels to call this
callback. For a TCP stream, this is always the last one called. For a HTTP
one, the callback is called when a request/response ends, so it is called
one time for each request/response.
* 'session_established' callback has been removed. Everything that is done in
this callback can be handled by 'channel_start_analyze' on the response
channel.
* 'http_pre_process' and 'http_post_process' callbacks have been replaced by
'channel_analyze'.
* 'http_start' callback has been replaced by 'http_headers'. This new one is
called just before headers sending and parsing of the body.
* 'http_end' callback has been replaced by 'channel_end_analyze'.
* It is possible to set a forwarder for TCP channels. It was already possible to
do it for HTTP ones.
* Forwarders can partially consumed forwardable data. For this reason a new
HTTP message state was added before HTTP_MSG_DONE : HTTP_MSG_ENDING.
Now all filters can define corresponding callbacks (http_forward_data
and tcp_forward_data). Each filter owns 2 offsets relative to buf->p, next and
forward, to track, respectively, input data already parsed but not forwarded yet
by the filter and parsed data considered as forwarded by the filter. A any time,
we have the warranty that a filter cannot parse or forward more input than
previous ones. And, of course, it cannot forward more input than it has
parsed. 2 macros has been added to retrieve these offets: FLT_NXT and FLT_FWD.
In addition, 2 functions has been added to change the 'next size' and the
'forward size' of a filter. When a filter parses input data, it can alter these
data, so the size of these data can vary. This action has an effet on all
previous filters that must be handled. To do so, the function
'filter_change_next_size' must be called, passing the size variation. In the
same spirit, if a filter alter forwarded data, it must call the function
'filter_change_forward_size'. 'filter_change_next_size' can be called in
'http_data' and 'tcp_data' callbacks and only these ones. And
'filter_change_forward_size' can be called in 'http_forward_data' and
'tcp_forward_data' callbacks and only these ones. The data changes are the
filter responsability, but with some limitation. It must not change already
parsed/forwarded data or data that previous filters have not parsed/forwarded
yet.
Because filters can be used on backends, when we the backend is set for a
stream, we add filters defined for this backend in the filter list of the
stream. But we must only do that when the backend and the frontend of the stream
are not the same. Else same filters are added a second time leading to undefined
behavior.
The HTTP compression code had to be moved.
So it simplifies http_response_forward_body function. To do so, the way the data
are forwarded has changed. Now, a filter (and only one) can forward data. In a
commit to come, this limitation will be removed to let all filters take part to
data forwarding. There are 2 new functions that filters should use to deal with
this feature:
* flt_set_http_data_forwarder: This function sets the filter (using its id)
that will forward data for the specified HTTP message. It is possible if it
was not already set by another filter _AND_ if no data was yet forwarded
(msg->msg_state <= HTTP_MSG_BODY). It returns -1 if an error occurs.
* flt_http_data_forwarder: This function returns the filter id that will
forward data for the specified HTTP message. If there is no forwarder set, it
returns -1.
When an HTTP data forwarder is set for the response, the HTTP compression is
disabled. Of course, this is not definitive.
diff --git a/Makefile b/Makefile
index d42d5fb..e6c94c8 100644
--- a/Makefile
+++ b/Makefile
@@ -750,7 +750,7 @@
src/session.o src/stream.o src/hdr_idx.o src/ev_select.o src/signal.o \
src/acl.o src/sample.o src/memory.o src/freq_ctr.o src/auth.o src/proto_udp.o \
src/compression.o src/payload.o src/hash.o src/pattern.o src/map.o \
- src/namespace.o src/mailers.o src/dns.o src/vars.o
+ src/namespace.o src/mailers.o src/dns.o src/vars.o src/filters.o
EBTREE_OBJS = $(EBTREE_DIR)/ebtree.o \
$(EBTREE_DIR)/eb32tree.o $(EBTREE_DIR)/eb64tree.o \
diff --git a/include/proto/filters.h b/include/proto/filters.h
new file mode 100644
index 0000000..d860424
--- /dev/null
+++ b/include/proto/filters.h
@@ -0,0 +1,138 @@
+/*
+ * include/proto/filters.h
+ * This file defines function prototypes for stream filters management.
+ *
+ * Copyright (C) 2015 Qualys Inc., Christopher Faulet <cfaulet@qualys.com>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation, version 2.1
+ * exclusively.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
+ */
+#ifndef _PROTO_FILTERS_H
+#define _PROTO_FILTERS_H
+
+#include <types/channel.h>
+#include <types/filters.h>
+#include <types/proto_http.h>
+#include <types/proxy.h>
+#include <types/stream.h>
+
+#include <proto/channel.h>
+
+/* Useful macros to access per-channel values. It can be safely used inside
+ * filters. */
+#define CHN_IDX(chn) (((chn)->flags & CF_ISRESP) == CF_ISRESP)
+#define FLT_NXT(flt, chn) ((flt)->next[CHN_IDX(chn)])
+#define FLT_FWD(flt, chn) ((flt)->fwd[CHN_IDX(chn)])
+
+extern struct pool_head *pool2_filter;
+
+int flt_init(struct proxy *p);
+void flt_deinit(struct proxy *p);
+int flt_check(struct proxy *p);
+
+int flt_stream_start(struct stream *s);
+void flt_stream_stop(struct stream *s);
+
+int flt_http_headers(struct stream *s, struct http_msg *msg);
+int flt_http_start_chunk(struct stream *s, struct http_msg *msg);
+int flt_http_data(struct stream *s, struct http_msg *msg);
+int flt_http_last_chunk(struct stream *s, struct http_msg *msg);
+int flt_http_end_chunk(struct stream *s, struct http_msg *msg);
+int flt_http_chunk_trailers(struct stream *s, struct http_msg *msg);
+int flt_http_end(struct stream *s, struct http_msg *msg);
+void flt_http_reset(struct stream *s, struct http_msg *msg);
+
+void flt_http_reply(struct stream *s, short status, const struct chunk *msg);
+int flt_http_forward_data(struct stream *s, struct http_msg *msg, unsigned int len);
+
+int flt_start_analyze(struct stream *s, struct channel *chn, unsigned int an_bit);
+int flt_analyze(struct stream *s, struct channel *chn, unsigned int an_bit);
+int flt_end_analyze(struct stream *s, struct channel *chn, unsigned int an_bit);
+
+int flt_xfer_data(struct stream *s, struct channel *chn, unsigned int an_bit);
+
+void flt_register_keywords(struct flt_kw_list *kwl);
+struct flt_kw *flt_find_kw(const char *kw);
+void flt_dump_kws(char **out);
+
+static inline void
+flt_set_forward_data(struct filter *filter, struct channel *chn)
+{
+ filter->flags[CHN_IDX(chn)] |= FILTER_FL_FORWARD_DATA;
+}
+
+static inline void
+flt_reset_forward_data(struct filter *filter, struct channel *chn)
+{
+ filter->flags[CHN_IDX(chn)] &= ~FILTER_FL_FORWARD_DATA;
+}
+
+static inline int
+flt_want_forward_data(struct filter *filter, const struct channel *chn)
+{
+ return filter->flags[CHN_IDX(chn)] & FILTER_FL_FORWARD_DATA;
+}
+
+
+/* This function must be called when a filter alter incoming data. It updates
+ * next offset value of all filter's predecessors. Do not call this function
+ * when a filter change the size of incomding data leads to an undefined
+ * behavior.
+ *
+ * This is the filter's responsiblitiy to update data itself. For now, it is
+ * unclear to know how to handle data updates, so we do the minimum here. For
+ * example, if you filter an HTTP message, we must update msg->next and
+ * msg->chunk_len values.
+ */
+static inline void
+flt_change_next_size(struct filter *filter, struct channel *chn, int len)
+{
+ struct stream *s = chn_strm(chn);
+ struct filter *f;
+
+ list_for_each_entry(f, &s->strm_flt.filters, list) {
+ if (f == filter)
+ break;
+ FLT_NXT(f, chn) += len;
+ }
+}
+
+/* This function must be called when a filter alter forwarded data. It updates
+ * offset values (next and forward) of all filters. Do not call this function
+ * when a filter change the size of forwarded data leads to an undefined
+ * behavior.
+ *
+ * This is the filter's responsiblitiy to update data itself. For now, it is
+ * unclear to know how to handle data updates, so we do the minimum here. For
+ * example, if you filter an HTTP message, we must update msg->next and
+ * msg->chunk_len values.
+ */
+static inline void
+flt_change_forward_size(struct filter *filter, struct channel *chn, int len)
+{
+ struct stream *s = chn_strm(chn);
+ struct filter *f;
+ int before = 1;
+
+ list_for_each_entry(f, &s->strm_flt.filters, list) {
+ if (f == filter)
+ before = 0;
+ if (before)
+ FLT_FWD(f, chn) += len;
+ FLT_NXT(f, chn) += len;
+ }
+}
+
+
+#endif /* _PROTO_FILTERS_H */
diff --git a/include/proto/proto_http.h b/include/proto/proto_http.h
index ef564dc..9317a55 100644
--- a/include/proto/proto_http.h
+++ b/include/proto/proto_http.h
@@ -126,6 +126,7 @@
void http_init_txn(struct stream *s);
void http_end_txn(struct stream *s);
void http_reset_txn(struct stream *s);
+void http_end_txn_clean_session(struct stream *s);
void http_adjust_conn_mode(struct stream *s, struct http_txn *txn, struct http_msg *msg);
struct act_rule *parse_http_req_cond(const char **args, const char *file, int linenum, struct proxy *proxy);
@@ -284,6 +285,7 @@
case HTTP_MSG_DATA: return "MSG_DATA";
case HTTP_MSG_CHUNK_CRLF: return "MSG_CHUNK_CRLF";
case HTTP_MSG_TRAILERS: return "MSG_TRAILERS";
+ case HTTP_MSG_ENDING: return "MSG_ENDING";
case HTTP_MSG_DONE: return "MSG_DONE";
case HTTP_MSG_CLOSING: return "MSG_CLOSING";
case HTTP_MSG_CLOSED: return "MSG_CLOSED";
diff --git a/include/types/channel.h b/include/types/channel.h
index e95c462..e43e8eb 100644
--- a/include/types/channel.h
+++ b/include/types/channel.h
@@ -157,6 +157,13 @@
#define AN_RES_STORE_RULES 0x00080000 /* table persistence matching */
#define AN_RES_HTTP_XFER_BODY 0x00100000 /* forward response body */
+#define AN_FLT_START_FE 0x01000000
+#define AN_FLT_START_BE 0x02000000
+#define AN_FLT_END 0x04000000
+#define AN_FLT_XFER_DATA 0x08000000
+
+#define AN_FLT_ALL_FE 0x0d000000
+#define AN_FLT_ALL_BE 0x0e000000
/* Magic value to forward infinite size (TCP, ...), used with ->to_forward */
#define CHN_INFINITE_FORWARD MAX_RANGE(unsigned int)
diff --git a/include/types/filters.h b/include/types/filters.h
new file mode 100644
index 0000000..635b2d1
--- /dev/null
+++ b/include/types/filters.h
@@ -0,0 +1,233 @@
+/*
+ * include/types/filteers.h
+ * This file defines everything related to stream filters.
+ *
+ * Copyright (C) 2015 Qualys Inc., Christopher Faulet <cfaulet@qualys.com>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation, version 2.1
+ * exclusively.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
+ */
+#ifndef _TYPES_FILTERS_H
+#define _TYPES_FILTERS_H
+
+#include <common/config.h>
+#include <common/mini-clist.h>
+
+struct http_msg;
+struct proxy;
+struct stream;
+struct channel;
+struct filter;
+
+/* Descriptor for a "filter" keyword. The ->parse() function returns 0 in case
+ * of success, or a combination of ERR_* flags if an error is encountered. The
+ * function pointer can be NULL if not implemented. The function also has an
+ * access to the current "server" config line. The ->skip value tells the parser
+ * how many words have to be skipped after the keyword. If the function needs to
+ * parse more keywords, it needs to update cur_arg.
+ */
+struct flt_kw {
+ const char *kw;
+ int (*parse)(char **args, int *cur_arg, struct proxy *px,
+ struct filter *filter, char **err);
+};
+
+/*
+ * A keyword list. It is a NULL-terminated array of keywords. It embeds a struct
+ * list in order to be linked to other lists, allowing it to easily be declared
+ * where it is needed, and linked without duplicating data nor allocating
+ * memory. It is also possible to indicate a scope for the keywords.
+ */
+struct flt_kw_list {
+ const char *scope;
+ struct list list;
+ struct flt_kw kw[VAR_ARRAY];
+};
+
+/*
+ * Filter flags set for a specific filter on channel
+ *
+ * - FILTER_FL_FORWARD_DATA : When this flag is set, the rest of the data is
+ * directly forwarded. For chunk-encoded HTTP
+ * messages, this flag is reseted between each
+ * chunks.
+ */
+#define FILTER_FL_FORWARD_DATA 0x00000001
+
+
+/*
+ * Callbacks available on a filter:
+ *
+ * - init : Initializes the filter for a proxy. Returns a
+ * negative value if an error occurs.
+ * - deinit : Cleans up what the init function has done.
+ * - check : Check the filter config for a proxy. Returns the
+ * number of errors encountered.
+ *
+ *
+ * - stream_start : Called when a stream is started. This callback will
+ * only be called for filters defined on a proxy with
+ * the frontend capability.
+ * Returns a negative value if an error occurs, any
+ * other value otherwise.
+ * - stream_stop : Called when a stream is stopped. This callback will
+ * only be called for filters defined on a proxy with
+ * the frontend capability.
+ *
+ *
+ * - channel_start_analyze: Called when a filter starts to analyze a channel.
+ * Returns a negative value if an error occurs, 0 if
+ * it needs to wait, any other value otherwise.
+ * - channel_analyze : Called before each analyzer attached to a channel,
+ * expects analyzers responsible for data sending.
+ * Returns a negative value if an error occurs, 0 if
+ * it needs to wait, any other value otherwise.
+ * - channel_end_analyze : Called when all other analyzers have finished their
+ * processing.
+ * Returns a negative value if an error occurs, 0 if
+ * it needs to wait, any other value otherwise.
+ *
+ *
+ * - http_headers : Called just before headers sending and parsing of
+ * the body. At this step, headers are fully parsed
+ * and the processing on it is finished.
+ * Returns a negative value if an error occurs, 0 if
+ * it needs to read more data (or to wait for some
+ * reason), any other value otherwise.
+ * - http_start_chunk : Called when we start to process a new chunk
+ * (for chunk-encoded request/response only). At this
+ * step, the chunk length is known and non-null.
+ * Returns a negative value if an error occurs, 0 if
+ * it needs to read more data (or to wait for some
+ * reason), any other value otherwise.
+ * - http_data : Called when unparsed body data are available.
+ * Returns a negative value if an error occurs, else
+ * the number of consumed bytes.
+ * - http_last_chunk : Called when the last chunk (with a zero length) is
+ * received.
+ * Returns a negative value if an error occurs, 0 if
+ * it needs to read more data (or to wait for some
+ * reason), any other value otherwise.
+ * - http_end_chunk : Called at the end of a chunk (expect for the last
+ * one).
+ * Returns a negative value if an error occurs, 0 if
+ * it needs to read more data (or to wait for some
+ * reason), any other value otherwise.
+ * - http_chunk_trailers : Called when part of trailer headers of a
+ * chunk-encoded request/response are ready to be
+ * processed.
+ * Returns a negative value if an error occurs, any
+ * other value otherwise.
+ * - http_end : Called when all the request/response has been
+ * processed and all body data has been forwarded.
+ * Returns a negative value if an error occurs, 0 if
+ * it needs to wait for some reason, any other value
+ * otherwise.
+ * - http_reset : Called when the HTTP message is reseted. It happens
+ * when a 100-continue response is received.
+ * Returns nothing.
+ * - http_reply : Called when, at any time, HA proxy decides to stop
+ * the HTTP message's processing and to send a message
+ * to the client (mainly, when an error or a redirect
+ * occur).
+ * Returns nothing.
+ * - http_forward_data : Called when some data can be consumed.
+ * Returns a negative value if an error occurs, else
+ * the number of forwarded bytes.
+ * - tcp_data : Called when unparsed data are available.
+ * Returns a negative value if an error occurs, else
+ * the number of consumed bytes.
+ * - tcp_forward_data : Called when some data can be consumed.
+ * Returns a negative value if an error occurs, else
+ * or the number of forwarded bytes.
+ */
+struct flt_ops {
+ /*
+ * Callbacks to manage the filter lifecycle
+ */
+ int (*init) (struct proxy *p, struct filter *f);
+ void (*deinit)(struct proxy *p, struct filter *f);
+ int (*check) (struct proxy *p, struct filter *f);
+
+ /*
+ * Stream callbacks
+ */
+ int (*stream_start) (struct stream *s, struct filter *f);
+ void (*stream_stop) (struct stream *s, struct filter *f);
+
+ /*
+ * Channel callbacks
+ */
+ int (*channel_start_analyze)(struct stream *s, struct filter *f, struct channel *chn);
+ int (*channel_analyze) (struct stream *s, struct filter *f, struct channel *chn, unsigned int an_bit);
+ int (*channel_end_analyze) (struct stream *s, struct filter *f, struct channel *chn);
+
+ /*
+ * HTTP callbacks
+ */
+ int (*http_headers) (struct stream *s, struct filter *f, struct http_msg *msg);
+ int (*http_start_chunk) (struct stream *s, struct filter *f, struct http_msg *msg);
+ int (*http_data) (struct stream *s, struct filter *f, struct http_msg *msg);
+ int (*http_last_chunk) (struct stream *s, struct filter *f, struct http_msg *msg);
+ int (*http_end_chunk) (struct stream *s, struct filter *f, struct http_msg *msg);
+ int (*http_chunk_trailers)(struct stream *s, struct filter *f, struct http_msg *msg);
+ int (*http_end) (struct stream *s, struct filter *f, struct http_msg *msg);
+ void (*http_reset) (struct stream *s, struct filter *f, struct http_msg *msg);
+
+ void (*http_reply) (struct stream *s, struct filter *f, short status,
+ const struct chunk *msg);
+ int (*http_forward_data) (struct stream *s, struct filter *f, struct http_msg *msg,
+ unsigned int len);
+
+ /*
+ * TCP callbacks
+ */
+ int (*tcp_data) (struct stream *s, struct filter *f, struct channel *chn);
+ int (*tcp_forward_data)(struct stream *s, struct filter *f, struct channel *chn,
+ unsigned int len);
+};
+
+/*
+ * Structure representing the state of a filter. When attached to a proxy, only
+ * <ops> and <conf> field (and optionnaly <id>) are set. All other fields are
+ * used when the filter is attached to a stream.
+ *
+ * 2D-Array fields are used to store info per channel. The first index stands
+ * for the request channel, and the second one for the response channel.
+ * Especially, <next> and <fwd> are offets representing amount of data that the
+ * filter are, respectively, parsed and forwarded on a channel. Filters can
+ * access these values using FLT_NXT and FLT_FWD macros.
+ */
+struct filter {
+ const char *id; /* The filter id */
+ struct flt_ops *ops; /* The filter callbacks */
+ void *conf; /* The filter configuration */
+ void *ctx; /* The filter context (opaque) */
+ int is_backend_filter; /* Flag to specify if the filter is a "backend" filter */
+ unsigned int flags[2]; /* 0: request, 1: response */
+ unsigned int next[2]; /* Offset, relative to buf->p, to the next byte to parse for a specific channel
+ * 0: request channel, 1: response channel */
+ unsigned int fwd[2]; /* Offset, relative to buf->p, to the next byte to forward for a specific channel
+ * 0: request channel, 1: response channel */
+ struct list list; /* Next filter for the same proxy/stream */
+};
+
+#endif /* _TYPES_FILTERS_H */
+
+/*
+ * Local variables:
+ * c-indent-level: 8
+ * c-basic-offset: 8
+ * End:
+ */
diff --git a/include/types/proto_http.h b/include/types/proto_http.h
index e5e9667..c6699bd 100644
--- a/include/types/proto_http.h
+++ b/include/types/proto_http.h
@@ -28,7 +28,7 @@
#include <common/regex.h>
#include <types/hdr_idx.h>
-#include <types/sample.h>
+#include <types/filters.h>
/* These are the flags that are found in txn->flags */
@@ -170,10 +170,11 @@
HTTP_MSG_CHUNK_CRLF = 31, // skipping CRLF after data chunk
HTTP_MSG_TRAILERS = 32, // trailers (post-data entity headers)
/* we enter this state when we've received the end of the current message */
- HTTP_MSG_DONE = 33, // message end received, waiting for resync or close
- HTTP_MSG_CLOSING = 34, // shutdown_w done, not all bytes sent yet
- HTTP_MSG_CLOSED = 35, // shutdown_w done, all bytes sent
- HTTP_MSG_TUNNEL = 36, // tunneled data after DONE
+ HTTP_MSG_ENDING = 33, // message end received, wait that the filters end too
+ HTTP_MSG_DONE = 34, // message end received, waiting for resync or close
+ HTTP_MSG_CLOSING = 35, // shutdown_w done, not all bytes sent yet
+ HTTP_MSG_CLOSED = 36, // shutdown_w done, all bytes sent
+ HTTP_MSG_TUNNEL = 37, // tunneled data after DONE
} __attribute__((packed));
/*
@@ -194,6 +195,7 @@
* contents if something needs them during a redispatch.
*/
#define HTTP_MSGF_WAIT_CONN 0x00000010 /* Wait for connect() to be confirmed before processing body */
+#define HTTP_MSGF_COMPRESSING 0x00000020 /* data compression is in progress */
/* Redirect flags */
diff --git a/include/types/proxy.h b/include/types/proxy.h
index e18ae72..71fd35d 100644
--- a/include/types/proxy.h
+++ b/include/types/proxy.h
@@ -38,6 +38,7 @@
#include <types/acl.h>
#include <types/backend.h>
#include <types/counters.h>
+#include <types/filters.h>
#include <types/freq_ctr.h>
#include <types/listener.h>
#include <types/log.h>
@@ -431,6 +432,8 @@
* this backend. If not specified or void, then the backend
* name is used
*/
+
+ struct list filters;
};
struct switching_rule {
diff --git a/include/types/stream.h b/include/types/stream.h
index bba5f43..292e36a 100644
--- a/include/types/stream.h
+++ b/include/types/stream.h
@@ -33,6 +33,7 @@
#include <types/channel.h>
#include <types/compression.h>
+#include <types/filters.h>
#include <types/hlua.h>
#include <types/obj_type.h>
#include <types/proto_http.h>
@@ -45,7 +46,6 @@
#include <types/stick_table.h>
#include <types/vars.h>
-
/* Various Stream Flags, bits values 0x01 to 0x100 (shift 0) */
#define SF_DIRECT 0x00000001 /* connection made on the server matching the client cookie */
#define SF_ASSIGNED 0x00000002 /* no need to assign a server to this stream */
@@ -127,6 +127,11 @@
struct http_txn *txn; /* current HTTP transaction being processed. Should become a list. */
+ struct {
+ struct list filters;
+ struct filter *current[2]; /* 0: request, 1: response */
+ } strm_flt;
+
struct task *task; /* the task associated with this stream */
struct list list; /* position in global streams list */
struct list by_srv; /* position in server stream list */
diff --git a/src/cfgparse.c b/src/cfgparse.c
index ec47542..22ff6af 100644
--- a/src/cfgparse.c
+++ b/src/cfgparse.c
@@ -45,6 +45,7 @@
#include <types/capture.h>
#include <types/compression.h>
+#include <types/filters.h>
#include <types/global.h>
#include <types/obj_type.h>
#include <types/peers.h>
@@ -58,6 +59,7 @@
#include <proto/checks.h>
#include <proto/compression.h>
#include <proto/dumpstats.h>
+#include <proto/filters.h>
#include <proto/frontend.h>
#include <proto/hdr_idx.h>
#include <proto/lb_chash.h>
@@ -8477,6 +8479,9 @@
}
}
+ /* Check filter configuration, if any */
+ cfgerr += flt_check(curproxy);
+
if (curproxy->cap & PR_CAP_FE) {
if (!curproxy->accept)
curproxy->accept = frontend_accept;
@@ -8492,6 +8497,12 @@
/* both TCP and HTTP must check switching rules */
curproxy->fe_req_ana |= AN_REQ_SWITCHING_RULES;
+
+ /* Add filters analyzers if needed */
+ if (!LIST_ISEMPTY(&curproxy->filters)) {
+ curproxy->fe_req_ana |= AN_FLT_ALL_FE;
+ curproxy->fe_rsp_ana |= AN_FLT_ALL_FE;
+ }
}
if (curproxy->cap & PR_CAP_BE) {
@@ -8512,6 +8523,12 @@
*/
if (curproxy->options2 & PR_O2_RDPC_PRST)
curproxy->be_req_ana |= AN_REQ_PRST_RDP_COOKIE;
+
+ /* Add filters analyzers if needed */
+ if (!LIST_ISEMPTY(&curproxy->filters)) {
+ curproxy->be_req_ana |= AN_FLT_ALL_BE;
+ curproxy->be_rsp_ana |= AN_FLT_ALL_BE;
+ }
}
}
diff --git a/src/compression.c b/src/compression.c
index 97f1fd7..471c102 100644
--- a/src/compression.c
+++ b/src/compression.c
@@ -218,11 +218,6 @@
/* restore original buffer pointer */
b_rew(in, msg->next);
-
- if (consumed_data > 0) {
- msg->next += consumed_data;
- msg->chunk_len -= consumed_data;
- }
return consumed_data;
}
@@ -307,7 +302,7 @@
if (msg->msg_state >= HTTP_MSG_TRAILERS) {
memcpy(tail, "0\r\n", 3);
tail += 3;
- if (msg->msg_state >= HTTP_MSG_DONE) {
+ if (msg->msg_state >= HTTP_MSG_ENDING) {
memcpy(tail, "\r\n", 2);
tail += 2;
}
diff --git a/src/filters.c b/src/filters.c
new file mode 100644
index 0000000..ab88f23
--- /dev/null
+++ b/src/filters.c
@@ -0,0 +1,946 @@
+/*
+ * Stream filters related variables and functions.
+ *
+ * Copyright (C) 2015 Qualys Inc., Christopher Faulet <cfaulet@qualys.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 <common/buffer.h>
+#include <common/debug.h>
+#include <common/cfgparse.h>
+#include <common/compat.h>
+#include <common/config.h>
+#include <common/errors.h>
+#include <common/namespace.h>
+#include <common/standard.h>
+
+#include <types/filters.h>
+#include <types/proto_http.h>
+
+#include <proto/compression.h>
+#include <proto/filters.h>
+#include <proto/proto_http.h>
+#include <proto/stream.h>
+#include <proto/stream_interface.h>
+
+/* Pool used to allocate filters */
+struct pool_head *pool2_filter = NULL;
+
+static int handle_analyzer_result(struct stream *s, struct channel *chn, unsigned int an_bit, int ret);
+
+/* - RESUME_FILTER_LOOP and RESUME_FILTER_END must always be used together.
+ * The first one begins a loop and the seconds one ends it.
+ *
+ * - BREAK_EXECUTION must be used to break the loop and set the filter from
+ * which to resume the next time.
+ *
+ * Here is an exemple:
+ *
+ * RESUME_FILTER_LOOP(stream, channel) {
+ * ...
+ * if (cond)
+ * BREAK_EXECUTION(stream, channel, label);
+ * ...
+ * } RESUME_FILTER_END;
+ * ...
+ * label:
+ * ...
+ *
+ */
+#define RESUME_FILTER_LOOP(strm, chn) \
+ do { \
+ struct filter *filter; \
+ \
+ if ((strm)->strm_flt.current[CHN_IDX(chn)]) { \
+ filter = (strm)->strm_flt.current[CHN_IDX(chn)]; \
+ (strm)->strm_flt.current[CHN_IDX(chn)] = NULL; \
+ goto resume_execution; \
+ } \
+ \
+ list_for_each_entry(filter, &s->strm_flt.filters, list) { \
+ resume_execution:
+
+#define RESUME_FILTER_END \
+ } \
+ } while(0)
+
+#define BREAK_EXECUTION(strm, chn, label) \
+ do { \
+ (strm)->strm_flt.current[CHN_IDX(chn)] = filter; \
+ goto label; \
+ } while (0)
+
+
+/* List head of all known filter keywords */
+static struct flt_kw_list flt_keywords = {
+ .list = LIST_HEAD_INIT(flt_keywords.list)
+};
+
+/*
+ * Registers the filter keyword list <kwl> as a list of valid keywords for next
+ * parsing sessions.
+ */
+void
+flt_register_keywords(struct flt_kw_list *kwl)
+{
+ LIST_ADDQ(&flt_keywords.list, &kwl->list);
+}
+
+/*
+ * Returns a pointer to the filter keyword <kw>, or NULL if not found. If the
+ * keyword is found with a NULL ->parse() function, then an attempt is made to
+ * find one with a valid ->parse() function. This way it is possible to declare
+ * platform-dependant, known keywords as NULL, then only declare them as valid
+ * if some options are met. Note that if the requested keyword contains an
+ * opening parenthesis, everything from this point is ignored.
+ */
+struct flt_kw *
+flt_find_kw(const char *kw)
+{
+ int index;
+ const char *kwend;
+ struct flt_kw_list *kwl;
+ struct flt_kw *ret = NULL;
+
+ kwend = strchr(kw, '(');
+ if (!kwend)
+ kwend = kw + strlen(kw);
+
+ list_for_each_entry(kwl, &flt_keywords.list, list) {
+ for (index = 0; kwl->kw[index].kw != NULL; index++) {
+ if ((strncmp(kwl->kw[index].kw, kw, kwend - kw) == 0) &&
+ kwl->kw[index].kw[kwend-kw] == 0) {
+ if (kwl->kw[index].parse)
+ return &kwl->kw[index]; /* found it !*/
+ else
+ ret = &kwl->kw[index]; /* may be OK */
+ }
+ }
+ }
+ return ret;
+}
+
+/*
+ * Dumps all registered "filter" keywords to the <out> string pointer. The
+ * unsupported keywords are only dumped if their supported form was not found.
+ */
+void
+flt_dump_kws(char **out)
+{
+ struct flt_kw_list *kwl;
+ int index;
+
+ *out = NULL;
+ list_for_each_entry(kwl, &flt_keywords.list, list) {
+ for (index = 0; kwl->kw[index].kw != NULL; index++) {
+ if (kwl->kw[index].parse ||
+ flt_find_kw(kwl->kw[index].kw) == &kwl->kw[index]) {
+ memprintf(out, "%s[%4s] %s%s\n", *out ? *out : "",
+ kwl->scope,
+ kwl->kw[index].kw,
+ kwl->kw[index].parse ? "" : " (not supported)");
+ }
+ }
+ }
+}
+
+/*
+ * Parses the "filter" keyword. All keywords must be handled by filters
+ * themselves
+ */
+static int
+parse_filter(char **args, int section_type, struct proxy *curpx,
+ struct proxy *defpx, const char *file, int line, char **err)
+{
+ struct filter *filter = NULL;
+
+ /* Filter cannot be defined on a default proxy */
+ if (curpx == defpx) {
+ memprintf(err, "parsing [%s:%d] : %s is only allowed in a 'default' section.",
+ file, line, args[0]);
+ return -1;
+ }
+ if (!strcmp(args[0], "filter")) {
+ struct flt_kw *kw;
+ int cur_arg;
+
+ if (!*args[1]) {
+ memprintf(err,
+ "parsing [%s:%d] : missing argument for '%s' in %s '%s'.",
+ file, line, args[0], proxy_type_str(curpx), curpx->id);
+ goto error;
+ }
+ filter = pool_alloc2(pool2_filter);
+ if (!filter) {
+ memprintf(err, "'%s' : out of memory", args[0]);
+ goto error;
+ }
+ memset(filter, 0, sizeof(*filter));
+
+ cur_arg = 1;
+ kw = flt_find_kw(args[cur_arg]);
+ if (kw) {
+ if (!kw->parse) {
+ memprintf(err, "parsing [%s:%d] : '%s' : "
+ "'%s' option is not implemented in this version (check build options).",
+ file, line, args[0], args[cur_arg]);
+ goto error;
+ }
+ if (kw->parse(args, &cur_arg, curpx, filter, err) != 0) {
+ if (err && *err)
+ memprintf(err, "'%s' : '%s'",
+ args[0], *err);
+ else
+ memprintf(err, "'%s' : error encountered while processing '%s'",
+ args[0], args[cur_arg]);
+ goto error;
+ }
+ }
+ else {
+ flt_dump_kws(err);
+ indent_msg(err, 4);
+ memprintf(err, "'%s' : unknown keyword '%s'.%s%s",
+ args[0], args[cur_arg],
+ err && *err ? " Registered keywords :" : "", err && *err ? *err : "");
+ goto error;
+ }
+ if (*args[cur_arg]) {
+ memprintf(err, "'%s %s' : unknown keyword '%s'.",
+ args[0], args[1], args[cur_arg]);
+ goto error;
+ }
+
+ LIST_ADDQ(&curpx->filters, &filter->list);
+ }
+ return 0;
+
+ error:
+ if (filter)
+ pool_free2(pool2_filter, filter);
+ return -1;
+
+
+}
+
+/*
+ * Calls 'init' callback for all filters attached to a proxy. This happens after
+ * the configuration parsing. Filters can finish to fill their config. Returns
+ * (ERR_ALERT|ERR_FATAL) if an error occurs, 0 otherwise.
+ */
+int
+flt_init(struct proxy *proxy)
+{
+ struct filter *filter;
+
+ list_for_each_entry(filter, &proxy->filters, list) {
+ if (filter->ops->init && filter->ops->init(proxy, filter) < 0)
+ return ERR_ALERT|ERR_FATAL;
+ }
+ return 0;
+}
+
+/*
+ * Calls 'check' callback for all filters attached to a proxy. This happens
+ * after the configuration parsing but before filters initialization. Returns
+ * the number of encountered errors.
+ */
+int
+flt_check(struct proxy *proxy)
+{
+ struct filter *filter;
+ int err = 0;
+
+ list_for_each_entry(filter, &proxy->filters, list) {
+ if (filter->ops->check)
+ err += filter->ops->check(proxy, filter);
+ }
+ return err;
+}
+
+/*
+ * Calls 'denit' callback for all filters attached to a proxy. This happens when
+ * HAProxy is stopped.
+ */
+void
+flt_deinit(struct proxy *proxy)
+{
+ struct filter *filter, *back;
+
+ list_for_each_entry_safe(filter, back, &proxy->filters, list) {
+ if (filter->ops->deinit)
+ filter->ops->deinit(proxy, filter);
+ LIST_DEL(&filter->list);
+ pool_free2(pool2_filter, filter);
+ }
+}
+
+/*
+ * Calls 'stream_start' for all filters attached to a stream. This happens when
+ * the stream is created, just after calling flt_stream_init
+ * function. Returns -1 if an error occurs, 0 otherwise.
+ */
+int
+flt_stream_start(struct stream *s)
+{
+ struct filter *filter;
+
+ list_for_each_entry(filter, &s->strm_flt.filters, list) {
+ if (filter->ops->stream_start && filter->ops->stream_start(s, filter) < 0)
+ return -1;
+ }
+ return 0;
+}
+
+/*
+ * Calls 'stream_stop' for all filters attached to a stream. This happens when
+ * the stream is stopped, just before calling flt_stream_release function.
+ */
+void
+flt_stream_stop(struct stream *s)
+{
+ struct filter *filter;
+
+ list_for_each_entry(filter, &s->strm_flt.filters, list) {
+ if (filter->ops->stream_stop)
+ filter->ops->stream_stop(s, filter);
+ }
+}
+
+int
+flt_http_headers(struct stream *s, struct http_msg *msg)
+{
+ struct filter *filter;
+ int ret = 1;
+
+ if (LIST_ISEMPTY(&s->strm_flt.filters))
+ goto end;
+
+ RESUME_FILTER_LOOP(s, msg->chn) {
+ if (filter->ops && filter->ops->http_headers) {
+ ret = filter->ops->http_headers(s, filter, msg);
+ if (ret <= 0)
+ BREAK_EXECUTION(s, msg->chn, end);
+ }
+ } RESUME_FILTER_END;
+
+ /* We increase FLT_NXT offset after all processing on headers because
+ * any filter can alter them. So the definitive size of headers
+ * (msg->sov) is only known when all filters have been called. */
+ list_for_each_entry(filter, &s->strm_flt.filters, list) {
+ FLT_NXT(filter, msg->chn) = msg->sov;
+ }
+ end:
+ return ret;
+}
+
+int
+flt_http_start_chunk(struct stream *s, struct http_msg *msg)
+{
+ int ret = 1;
+
+ if (LIST_ISEMPTY(&s->strm_flt.filters))
+ goto end;
+
+ RESUME_FILTER_LOOP(s, msg->chn) {
+ if (filter->ops->http_start_chunk) {
+ ret = filter->ops->http_start_chunk(s, filter, msg);
+ if (ret <= 0)
+ BREAK_EXECUTION(s, msg->chn, end);
+ }
+ FLT_NXT(filter, msg->chn) += msg->sol;
+ } RESUME_FILTER_END;
+ end:
+ return ret;
+}
+
+/*
+ * Calls 'http_data' callback for all "data" filters attached to a stream. This
+ * function is called when incoming data are available (excluding chunks
+ * envelope for chunked messages) in the AN_REQ_HTTP_XFER_BODY and
+ * AN_RES_HTTP_XFER_BODY analyzers. It takes care to update the next offset of
+ * filters and adjusts available data to be sure that a filter cannot parse more
+ * data than its predecessors. A filter can choose to not consume all available
+ * data. Returns -1 if an error occurs, the number of consumed bytes otherwise.
+ */
+int
+flt_http_data(struct stream *s, struct http_msg *msg)
+{
+ struct filter *filter = NULL;
+ unsigned int buf_i;
+ int ret = 0;
+
+ /* No filter, consume all available data */
+ if (LIST_ISEMPTY(&s->strm_flt.filters)) {
+ ret = MIN(msg->chunk_len, msg->chn->buf->i - msg->next);
+ goto end;
+ }
+
+ /* Save buffer state */
+ buf_i = msg->chn->buf->i;
+ list_for_each_entry(filter, &s->strm_flt.filters, list) {
+ if (filter->ops->http_data && !flt_want_forward_data(filter, msg->chn)) {
+ ret = filter->ops->http_data(s, filter, msg);
+ if (ret < 0)
+ break;
+ }
+ else {
+ /* msg->chunk_len is the remaining size of data to parse
+ * in the body (or in the current chunk for
+ * chunk-encoded messages) from the HTTP parser point of
+ * view (relatively to msg->next). To have it from the
+ * filter point of view, we need to be add (msg->next
+ * -FLT_NEXT) to it. */
+ ret = MIN(msg->chunk_len + msg->next, msg->chn->buf->i) - FLT_NXT(filter, msg->chn);
+ }
+
+ /* Increase FLT_NXT offset of the current filter */
+ FLT_NXT(filter, msg->chn) += ret;
+
+ /* And set this value as the bound for the next filter. It will
+ * not able to parse more data than the current one. */
+ msg->chn->buf->i = FLT_NXT(filter, msg->chn);
+ }
+ /* Restore the original buffer state */
+ msg->chn->buf->i = buf_i;
+ end:
+ return ret;
+}
+
+int
+flt_http_end_chunk(struct stream *s, struct http_msg *msg)
+{
+ int ret = 1;
+
+ if (LIST_ISEMPTY(&s->strm_flt.filters))
+ goto end;
+
+ RESUME_FILTER_LOOP(s, msg->chn) {
+ if (filter->ops->http_end_chunk) {
+ ret = filter->ops->http_end_chunk(s, filter, msg);
+ if (ret <= 0)
+ BREAK_EXECUTION(s, msg->chn, end);
+ }
+ flt_reset_forward_data(filter, msg->chn);
+ FLT_NXT(filter, msg->chn) += msg->sol;
+ } RESUME_FILTER_END;
+ end:
+ return ret;
+}
+
+int
+flt_http_last_chunk(struct stream *s, struct http_msg *msg)
+{
+ int ret = 1;
+
+ if (LIST_ISEMPTY(&s->strm_flt.filters))
+ goto end;
+
+ RESUME_FILTER_LOOP(s, msg->chn) {
+ if (filter->ops->http_last_chunk) {
+ ret = filter->ops->http_last_chunk(s, filter, msg);
+ if (ret <= 0)
+ BREAK_EXECUTION(s, msg->chn, end);
+ }
+ flt_reset_forward_data(filter, msg->chn);
+ FLT_NXT(filter, msg->chn) += msg->sol;
+ } RESUME_FILTER_END;
+ end:
+ return ret;
+}
+
+
+/*
+ * Calls 'http_chunk_trailers' callback for all "data" filters attached to a
+ * stream. This function is called for chunked messages only when a part of the
+ * trailers was parsed in the AN_REQ_HTTP_XFER_BODY and AN_RES_HTTP_XFER_BODY
+ * analyzers. Filters can know how much data were parsed by the HTTP parsing
+ * until the last call with the msg->sol value. Returns a negative value if an
+ * error occurs, any other value otherwise.
+ */
+int
+flt_http_chunk_trailers(struct stream *s, struct http_msg *msg)
+{
+ int ret = 1;
+
+ if (LIST_ISEMPTY(&s->strm_flt.filters))
+ goto end;
+
+ RESUME_FILTER_LOOP(s, msg->chn) {
+ if (filter->ops->http_chunk_trailers) {
+ ret = filter->ops->http_chunk_trailers(s, filter, msg);
+ if (ret <= 0)
+ BREAK_EXECUTION(s, msg->chn, end);
+ }
+ FLT_NXT(filter, msg->chn) += msg->sol;
+ } RESUME_FILTER_END;
+end:
+ return ret;
+}
+
+/*
+ * Calls 'http_end' callback for all filters attached to a stream. All filters
+ * are called here, but only if there is at least one "data" filter. This
+ * functions is called when all data were parsed and forwarded. 'http_end'
+ * callback is resumable, so this function returns a negative value if an error
+ * occurs, 0 if it needs to wait for some reason, any other value otherwise.
+ */
+int
+flt_http_end(struct stream *s, struct http_msg *msg)
+{
+ int ret = 1;
+
+ if (LIST_ISEMPTY(&s->strm_flt.filters))
+ goto end;
+
+ RESUME_FILTER_LOOP(s, msg->chn) {
+ if (filter->ops->http_end) {
+ ret = filter->ops->http_end(s, filter, msg);
+ if (ret <= 0)
+ BREAK_EXECUTION(s, msg->chn, end);
+ }
+ flt_reset_forward_data(filter, msg->chn);
+ } RESUME_FILTER_END;
+end:
+ return ret;
+}
+
+/*
+ * Calls 'http_reset' callback for all filters attached to a stream. This
+ * happens when a 100-continue response is received.
+ */
+void
+flt_http_reset(struct stream *s, struct http_msg *msg)
+{
+ struct filter *filter;
+
+ if (LIST_ISEMPTY(&s->strm_flt.filters))
+ return;
+
+ list_for_each_entry(filter, &s->strm_flt.filters, list) {
+ if (filter->ops->http_reset)
+ filter->ops->http_reset(s, filter, msg);
+ }
+}
+
+/*
+ * Calls 'http_reply' callback for all filters attached to a stream when HA
+ * decides to stop the HTTP message processing.
+ */
+void
+flt_http_reply(struct stream *s, short status, const struct chunk *msg)
+{
+ struct filter *filter;
+
+ if (LIST_ISEMPTY(&s->strm_flt.filters))
+ return;
+
+ list_for_each_entry(filter, &s->strm_flt.filters, list) {
+ if (filter->ops->http_reply)
+ filter->ops->http_reply(s, filter, status, msg);
+ }
+}
+
+/*
+ * Calls 'http_forward_data' callback for all "data" filters attached to a
+ * stream. This function is called when some data can be forwarded in the
+ * AN_REQ_HTTP_XFER_BODY and AN_RES_HTTP_XFER_BODY analyzers. It takes care to
+ * update the forward offset of filters and adjusts "forwardable" data to be
+ * sure that a filter cannot forward more data than its predecessors. A filter
+ * can choose to not forward all parsed data. Returns a negative value if an
+ * error occurs, else the number of forwarded bytes.
+ */
+int
+flt_http_forward_data(struct stream *s, struct http_msg *msg, unsigned int len)
+{
+ struct filter *filter = NULL;
+ int ret = len;
+
+ /* No filter, forward all data */
+ if (LIST_ISEMPTY(&s->strm_flt.filters))
+ goto end;
+
+ list_for_each_entry(filter, &s->strm_flt.filters, list) {
+ if (filter->ops->http_forward_data) {
+ /* Remove bytes that the current filter considered as
+ * forwarded */
+ ret = filter->ops->http_forward_data(s, filter, msg,
+ ret - FLT_FWD(filter, msg->chn));
+ if (ret < 0)
+ goto end;
+ }
+
+ /* Adjust bytes that the current filter considers as
+ * forwarded */
+ FLT_FWD(filter, msg->chn) += ret;
+
+ /* And set this value as the bound for the next filter. It will
+ * not able to forward more data than the current one. */
+ ret = FLT_FWD(filter, msg->chn);
+ }
+
+ if (!ret)
+ goto end;
+
+ /* Finally, adjust filters offsets by removing data that HAProxy will
+ * forward. */
+ list_for_each_entry(filter, &s->strm_flt.filters, list) {
+ FLT_NXT(filter, msg->chn) -= ret;
+ FLT_FWD(filter, msg->chn) -= ret;
+ }
+ end:
+ return ret;
+}
+
+/*
+ * Calls 'channel_start_analyze' callback for all filters attached to a
+ * stream. This function is called when we start to analyze a request or a
+ * response. For frontend filters, it is called before all other analyzers. For
+ * backend ones, it is called before all backend
+ * analyzers. 'channel_start_analyze' callback is resumable, so this function
+ * returns 0 if an error occurs or if it needs to wait, any other value
+ * otherwise.
+ */
+int
+flt_start_analyze(struct stream *s, struct channel *chn, unsigned int an_bit)
+{
+ int ret = 1;
+
+ /* If this function is called, this means there is at least one filter,
+ * so we do not need to check the filter list's emptiness. */
+
+ RESUME_FILTER_LOOP(s, chn) {
+ if (an_bit == AN_FLT_START_BE && !filter->is_backend_filter)
+ continue;
+
+ filter->next[CHN_IDX(chn)] = 0;
+ filter->fwd[CHN_IDX(chn)] = 0;
+
+ if (filter->ops->channel_start_analyze) {
+ ret = filter->ops->channel_start_analyze(s, filter, chn);
+ if (ret <= 0)
+ BREAK_EXECUTION(s, chn, end);
+ }
+ } RESUME_FILTER_END;
+
+ end:
+ return handle_analyzer_result(s, chn, an_bit, ret);
+}
+
+/*
+ * Calls 'channel_analyze' callback for all filters attached to a stream. This
+ * function is called before each analyzer attached to a channel, expects
+ * analyzers responsible for data sending. 'channel_analyze' callback is
+ * resumable, so this function returns 0 if an error occurs or if it needs to
+ * wait, any other value otherwise.
+ */
+int
+flt_analyze(struct stream *s, struct channel *chn, unsigned int an_bit)
+{
+ int ret = 1;
+
+ if (LIST_ISEMPTY(&s->strm_flt.filters))
+ goto end;
+
+ RESUME_FILTER_LOOP(s, chn) {
+ if (filter->ops->channel_analyze) {
+ ret = filter->ops->channel_analyze(s, filter, chn, an_bit);
+ if (ret <= 0)
+ BREAK_EXECUTION(s, chn, check_result);
+ }
+ } RESUME_FILTER_END;
+
+ check_result:
+ ret = handle_analyzer_result(s, chn, 0, ret);
+ end:
+ return ret;
+}
+
+/*
+ * Calls 'channel_end_analyze' callback for all filters attached to a
+ * stream. This function is called when we stop to analyze a request or a
+ * response. It is called after all other analyzers. 'channel_end_analyze'
+ * callback is resumable, so this function returns 0 if an error occurs or if it
+ * needs to wait, any other value otherwise.
+ */
+int
+flt_end_analyze(struct stream *s, struct channel *chn, unsigned int an_bit)
+{
+ int ret = 1;
+
+ /* If this function is called, this means there is at least one filter,
+ * so we do not need to check the filter list's emptiness. */
+
+ RESUME_FILTER_LOOP(s, chn) {
+ filter->next[CHN_IDX(chn)] = 0;
+
+ if (filter->ops->channel_end_analyze) {
+ ret = filter->ops->channel_end_analyze(s, filter, chn);
+ if (ret <= 0)
+ BREAK_EXECUTION(s, chn, end);
+ }
+ } RESUME_FILTER_END;
+
+end:
+ ret = handle_analyzer_result(s, chn, an_bit, ret);
+ if (!(s->req.analysers & AN_FLT_END) &&
+ !(s->res.analysers & AN_FLT_END) &&
+ s->txn && (s->txn->flags & TX_WAIT_NEXT_RQ)) {
+ struct filter *filter, *back;
+
+ s->req.analysers = strm_li(s) ? strm_li(s)->analysers : 0;
+ s->res.analysers = 0;
+
+ list_for_each_entry_safe(filter, back, &s->strm_flt.filters, list) {
+ if (filter->is_backend_filter) {
+ LIST_DEL(&filter->list);
+ pool_free2(pool2_filter, filter);
+ }
+ }
+ }
+ else if (ret) {
+ /* Analyzer ends only for one channel. So wake up the stream to
+ * be sure to process it for the other side as soon as
+ * possible. */
+ task_wakeup(s->task, TASK_WOKEN_MSG);
+ }
+ return ret;
+}
+
+
+/*
+ * Calls 'tcp_data' callback for all "data" filters attached to a stream. This
+ * function is called when incoming data are available. It takes care to update
+ * the next offset of filters and adjusts available data to be sure that a
+ * filter cannot parse more data than its predecessors. A filter can choose to
+ * not consume all available data. Returns -1 if an error occurs, the number of
+ * consumed bytes otherwise.
+ */
+static int
+flt_data(struct stream *s, struct channel *chn)
+{
+ struct filter *filter = NULL;
+ unsigned int buf_i;
+ int ret = chn->buf->i;
+
+ /* Save buffer state */
+ buf_i = chn->buf->i;
+ list_for_each_entry(filter, &s->strm_flt.filters, list) {
+ if (filter->ops->tcp_data && !flt_want_forward_data(filter, chn)) {
+ ret = filter->ops->tcp_data(s, filter, chn);
+ if (ret < 0)
+ break;
+ }
+ else
+ ret = chn->buf->i - FLT_NXT(filter, chn);
+
+ /* Increase next offset of the current filter */
+ FLT_NXT(filter, chn) += ret;
+
+ /* Update <ret> value to be sure to have the last one when we
+ * exit from the loop. */
+ ret = FLT_NXT(filter, chn);
+
+ /* And set this value as the bound for the next filter. It will
+ * not able to parse more data than the current one. */
+ chn->buf->i = FLT_NXT(filter, chn);
+ }
+ // Restore the original buffer state
+ chn->buf->i = buf_i;
+ return ret;
+}
+
+/*
+ * Calls 'tcp_forward_data' callback for all "data" filters attached to a
+ * stream. This function is called when some data can be forwarded. It takes
+ * care to update the forward offset of filters and adjusts "forwardable" data
+ * to be sure that a filter cannot forward more data than its predecessors. A
+ * filter can choose to not forward all parsed data. Returns a negative value if
+ * an error occurs, else the number of forwarded bytes.
+ */
+static int
+flt_forward_data(struct stream *s, struct channel *chn, unsigned int len)
+{
+ struct filter *filter = NULL;
+ int ret = len;
+
+ list_for_each_entry(filter, &s->strm_flt.filters, list) {
+ if (filter->ops->tcp_forward_data) {
+ /* Remove bytes that the current filter considered as
+ * forwarded */
+ ret = filter->ops->tcp_forward_data(s, filter, chn, ret - FLT_FWD(filter, chn));
+ if (ret < 0)
+ goto end;
+ }
+
+ /* Adjust bytes taht the current filter considers as
+ * forwarded */
+ FLT_FWD(filter, chn) += ret;
+
+ /* And set this value as the bound for the next filter. It will
+ * not able to forward more data than the current one. */
+ ret = FLT_FWD(filter, chn);
+ }
+
+ if (!ret)
+ goto end;
+
+ /* Adjust forward counter and next offset of filters by removing data
+ * that HAProxy will consider as forwarded. */
+ list_for_each_entry(filter, &s->strm_flt.filters, list) {
+ FLT_NXT(filter, chn) -= ret;
+ FLT_FWD(filter, chn) -= ret;
+ }
+
+ /* Consume data that all filters consider as forwarded. */
+ b_adv(chn->buf, ret);
+ end:
+ return ret;
+}
+
+/*
+ * Called when TCP data must be filtered on a channel. This function is the
+ * AN_FLT_XFER_DATA analyzer. When called, it is responsible to forward data
+ * when the proxy is not in http mode. Behind the scene, it calls consecutively
+ * 'tcp_data' and 'tcp_forward_data' callbacks for all "data" filters attached
+ * to a stream. Returns 0 if an error occurs or if it needs to wait, any other
+ * value otherwise.
+ */
+int
+flt_xfer_data(struct stream *s, struct channel *chn, unsigned int an_bit)
+{
+ int ret = 1;
+
+ /* If this function is called, this means there is at least one filter,
+ * so we do not need to check the filter list's emptiness. */
+
+ /* Be sure that the output is still opened. Else we stop the data
+ * filtering. */
+ if ((chn->flags & (CF_READ_ERROR|CF_READ_TIMEOUT|CF_WRITE_ERROR|CF_WRITE_TIMEOUT)) ||
+ ((chn->flags & CF_SHUTW) && (chn->to_forward || chn->buf->o)))
+ goto end;
+
+ /* Let all "data" filters parsing incoming data */
+ ret = flt_data(s, chn);
+ if (ret < 0)
+ goto end;
+
+ /* And forward them */
+ ret = flt_forward_data(s, chn, ret);
+ if (ret < 0)
+ goto end;
+
+ /* Stop waiting data if the input in closed and no data is pending or if
+ * the output is closed. */
+ if ((chn->flags & CF_SHUTW) ||
+ ((chn->flags & CF_SHUTR) && !buffer_pending(chn->buf))) {
+ ret = 1;
+ goto end;
+ }
+
+ /* Wait for data */
+ return 0;
+ end:
+ /* Terminate the data filtering. If <ret> is negative, an error was
+ * encountered during the filtering. */
+ return handle_analyzer_result(s, chn, an_bit, ret);
+}
+
+/*
+ * Handles result of filter's analyzers. It returns 0 if an error occurs or if
+ * it needs to wait, any other value otherwise.
+ */
+static int
+handle_analyzer_result(struct stream *s, struct channel *chn,
+ unsigned int an_bit, int ret)
+{
+ int finst;
+
+ if (ret < 0)
+ goto return_bad_req;
+ else if (!ret)
+ goto wait;
+
+ /* End of job, return OK */
+ if (an_bit) {
+ chn->analysers &= ~an_bit;
+ chn->analyse_exp = TICK_ETERNITY;
+ }
+ return 1;
+
+ return_bad_req:
+ /* An error occurs */
+ channel_abort(&s->req);
+ channel_abort(&s->res);
+
+ if (!(chn->flags & CF_ISRESP)) {
+ s->req.analysers &= AN_FLT_END;
+ finst = SF_FINST_R;
+ /* FIXME: incr counters */
+ }
+ else {
+ s->res.analysers &= AN_FLT_END;
+ finst = SF_FINST_H;
+ /* FIXME: incr counters */
+ }
+
+ if (s->txn) {
+ /* Do not do that when we are waiting for the next request */
+ if (s->txn->status)
+ http_reply_and_close(s, s->txn->status, NULL);
+ else {
+ s->txn->status = 400;
+ http_reply_and_close(s, 400, http_error_message(s, HTTP_ERR_400));
+ }
+ }
+
+ if (!(s->flags & SF_ERR_MASK))
+ s->flags |= SF_ERR_PRXCOND;
+ if (!(s->flags & SF_FINST_MASK))
+ s->flags |= finst;
+ return 0;
+
+ wait:
+ if (!(chn->flags & CF_ISRESP))
+ channel_dont_connect(chn);
+ return 0;
+}
+
+
+/* Note: must not be declared <const> as its list will be overwritten.
+ * Please take care of keeping this list alphabetically sorted, doing so helps
+ * all code contributors.
+ * Optional keywords are also declared with a NULL ->parse() function so that
+ * the config parser can report an appropriate error when a known keyword was
+ * not enabled. */
+static struct cfg_kw_list cfg_kws = {ILH, {
+ { CFG_LISTEN, "filter", parse_filter },
+ { 0, NULL, NULL },
+ }
+};
+
+__attribute__((constructor))
+static void
+__filters_init(void)
+{
+ pool2_filter = create_pool("filter", sizeof(struct filter), MEM_F_SHARED);
+ cfg_register_keywords(&cfg_kws);
+}
+
+__attribute__((destructor))
+static void
+__filters_deinit(void)
+{
+ pool_destroy2(pool2_filter);
+}
+
+/*
+ * Local variables:
+ * c-indent-level: 8
+ * c-basic-offset: 8
+ * End:
+ */
diff --git a/src/haproxy.c b/src/haproxy.c
index f7b4ced..8ffdb67 100644
--- a/src/haproxy.c
+++ b/src/haproxy.c
@@ -76,6 +76,7 @@
#include <common/version.h>
#include <types/capture.h>
+#include <types/filters.h>
#include <types/global.h>
#include <types/acl.h>
#include <types/peers.h>
@@ -89,6 +90,7 @@
#include <proto/checks.h>
#include <proto/connection.h>
#include <proto/fd.h>
+#include <proto/filters.h>
#include <proto/hdr_idx.h>
#include <proto/hlua.h>
#include <proto/listener.h>
@@ -560,6 +562,7 @@
char *progname;
char *change_dir = NULL;
struct tm curtime;
+ struct proxy *px;
chunk_init(&trash, malloc(global.tune.bufsize), global.tune.bufsize);
alloc_trash_buffers(global.tune.bufsize);
@@ -861,6 +864,15 @@
init_51degrees();
#endif
+ for (px = proxy; px; px = px->next) {
+ err_code |= flt_init(px);
+ if (err_code & (ERR_ABORT|ERR_FATAL)) {
+ Alert("Failed to initialize filters for proxy '%s'.\n",
+ px->id);
+ exit(1);
+ }
+ }
+
if (start_checks() < 0)
exit(1);
@@ -1468,6 +1480,8 @@
free(bind_conf);
}
+ flt_deinit(p);
+
free(p->desc);
free(p->fwdfor_hdr_name);
@@ -1550,7 +1564,6 @@
pool_destroy2(pool2_sig_handlers);
pool_destroy2(pool2_hdr_idx);
pool_destroy2(pool2_http_txn);
-
deinit_pollers();
} /* end deinit() */
diff --git a/src/proto_http.c b/src/proto_http.c
index 247c3b6..79b21f9 100644
--- a/src/proto_http.c
+++ b/src/proto_http.c
@@ -39,6 +39,7 @@
#include <common/version.h>
#include <types/capture.h>
+#include <types/filters.h>
#include <types/global.h>
#include <proto/acl.h>
@@ -51,6 +52,7 @@
#include <proto/compression.h>
#include <proto/dumpstats.h>
#include <proto/fd.h>
+#include <proto/filters.h>
#include <proto/frontend.h>
#include <proto/log.h>
#include <proto/hdr_idx.h>
@@ -933,6 +935,7 @@
static void http_server_error(struct stream *s, struct stream_interface *si,
int err, int finst, int status, const struct chunk *msg)
{
+ flt_http_reply(s, status, msg);
channel_auto_read(si_oc(si));
channel_abort(si_oc(si));
channel_auto_close(si_oc(si));
@@ -966,6 +969,8 @@
void
http_reply_and_close(struct stream *s, short status, struct chunk *msg)
{
+ s->txn->flags &= ~TX_WAIT_NEXT_RQ;
+ flt_http_reply(s, status, msg);
stream_int_retnclose(&s->si[0], msg);
}
@@ -2180,10 +2185,8 @@
msg->sol = ptr - ptr_old;
if (unlikely(ptr < ptr_old))
msg->sol += buf->size;
- msg->next = buffer_count(buf, buf->p, ptr);
msg->chunk_len = chunk;
msg->body_len += chunk;
- msg->msg_state = chunk ? HTTP_MSG_DATA : HTTP_MSG_TRAILERS;
return 1;
error:
msg->err_pos = buffer_count(buf, buf->p, ptr);
@@ -2211,7 +2214,7 @@
/* we have msg->next which points to next line. Look for CRLF. */
while (1) {
const char *p1 = NULL, *p2 = NULL;
- const char *ptr = b_ptr(buf, msg->next);
+ const char *ptr = b_ptr(buf, msg->next + msg->sol);
const char *stop = bi_end(buf);
int bytes;
@@ -2245,21 +2248,20 @@
if (p2 >= buf->data + buf->size)
p2 = buf->data;
- bytes = p2 - b_ptr(buf, msg->next);
+ bytes = p2 - b_ptr(buf, msg->next + msg->sol);
if (bytes < 0)
bytes += buf->size;
- if (p1 == b_ptr(buf, msg->next)) {
- /* LF/CRLF at beginning of line => end of trailers at p2.
- * Everything was scheduled for forwarding, there's nothing
- * left from this message.
- */
- msg->next = buffer_count(buf, buf->p, p2);
- msg->msg_state = HTTP_MSG_DONE;
+ /* LF/CRLF at beginning of line => end of trailers at p2.
+ * Everything was scheduled for forwarding, there's nothing left
+ * from this message.
+ */
+ if (p1 == b_ptr(buf, msg->next + msg->sol)) {
+ msg->sol += bytes;
return 1;
}
+ msg->sol += bytes;
/* OK, next line then */
- msg->next = buffer_count(buf, buf->p, p2);
}
}
@@ -2299,13 +2301,7 @@
msg->err_pos = buffer_count(buf, buf->p, ptr);
return -1;
}
-
- ptr++;
- if (unlikely(ptr >= buf->data + buf->size))
- ptr = buf->data;
- /* Advance ->next to allow the CRLF to be forwarded */
- msg->next += bytes;
- msg->msg_state = HTTP_MSG_CHUNK_SIZE;
+ msg->sol = bytes;
return 1;
}
@@ -2475,11 +2471,12 @@
/* compress 200,201,202,203 responses only */
if ((txn->status != 200) &&
- (txn->status != 201) &&
- (txn->status != 202) &&
- (txn->status != 203))
+ (txn->status != 201) &&
+ (txn->status != 202) &&
+ (txn->status != 203))
goto fail;
+
/* Content-Length is null */
if (!(msg->flags & HTTP_MSGF_TE_CHNK) && msg->body_len == 0)
goto fail;
@@ -2812,8 +2809,7 @@
txn->status = 400;
msg->msg_state = HTTP_MSG_ERROR;
http_reply_and_close(s, txn->status, NULL);
- req->analysers = 0;
-
+ req->analysers &= AN_FLT_END;
stream_inc_http_req_ctr(s);
proxy_inc_fe_req_ctr(sess->fe);
sess->fe->fe_counters.failed_req++;
@@ -2844,7 +2840,7 @@
txn->status = 408;
msg->msg_state = HTTP_MSG_ERROR;
http_reply_and_close(s, txn->status, http_error_message(s, HTTP_ERR_408));
- req->analysers = 0;
+ req->analysers &= AN_FLT_END;
stream_inc_http_req_ctr(s);
proxy_inc_fe_req_ctr(sess->fe);
@@ -2873,8 +2869,7 @@
txn->status = 400;
msg->msg_state = HTTP_MSG_ERROR;
http_reply_and_close(s, txn->status, http_error_message(s, HTTP_ERR_400));
- req->analysers = 0;
-
+ req->analysers &= AN_FLT_END;
stream_inc_http_err_ctr(s);
stream_inc_http_req_ctr(s);
proxy_inc_fe_req_ctr(sess->fe);
@@ -2930,7 +2925,7 @@
*/
txn->status = 0;
msg->msg_state = HTTP_MSG_RQBEFORE;
- req->analysers = 0;
+ req->analysers &= AN_FLT_END;
s->logs.logwait = 0;
s->logs.level = 0;
s->res.flags &= ~CF_EXPECT_MORE; /* speed up sending a previous response */
@@ -3257,7 +3252,7 @@
if (!(s->flags & SF_FINST_MASK))
s->flags |= SF_FINST_R;
- req->analysers = 0;
+ req->analysers &= AN_FLT_END;
req->analyse_exp = TICK_ETERNITY;
return 0;
}
@@ -4260,13 +4255,14 @@
}
memcpy(trash.str + trash.len, "\r\n\r\n", 4);
trash.len += 4;
+ flt_http_reply(s, txn->status, &trash);
bo_inject(res->chn, trash.str, trash.len);
/* "eat" the request */
bi_fast_delete(req->chn->buf, req->sov);
req->next -= req->sov;
req->sov = 0;
- s->req.analysers = AN_REQ_HTTP_XFER_BODY;
- s->res.analysers = AN_RES_HTTP_XFER_BODY;
+ s->req.analysers = AN_REQ_HTTP_XFER_BODY | (s->req.analysers & AN_FLT_END);
+ s->res.analysers = AN_RES_HTTP_XFER_BODY | (s->req.analysers & AN_FLT_END);
req->msg_state = HTTP_MSG_CLOSED;
res->msg_state = HTTP_MSG_DONE;
/* Trim any possible response */
@@ -4282,7 +4278,7 @@
trash.len += 23;
}
http_reply_and_close(s, txn->status, &trash);
- req->chn->analysers = 0;
+ req->chn->analysers &= AN_FLT_END;
}
if (!(s->flags & SF_ERR_MASK))
@@ -4428,7 +4424,9 @@
select_compression_request_header(s, req->buf);
/* enable the minimally required analyzers to handle keep-alive and compression on the HTTP response */
- req->analysers = (req->analysers & AN_REQ_HTTP_BODY) | AN_REQ_HTTP_XFER_BODY;
+ req->analysers &= (AN_REQ_HTTP_BODY | AN_FLT_END);
+ req->analysers &= ~AN_FLT_XFER_DATA;
+ req->analysers |= AN_REQ_HTTP_XFER_BODY;
goto done;
}
@@ -4478,7 +4476,7 @@
* if the client closes first.
*/
channel_dont_connect(req);
- req->analysers = 0; /* remove switching rules etc... */
+ req->analysers &= AN_FLT_END; /* remove switching rules etc... */
req->analysers |= AN_REQ_HTTP_TARPIT;
req->analyse_exp = tick_add_ifset(now_ms, s->be->timeout.tarpit);
if (!req->analyse_exp)
@@ -4527,7 +4525,7 @@
if (!(s->flags & SF_FINST_MASK))
s->flags |= SF_FINST_R;
- req->analysers = 0;
+ req->analysers &= AN_FLT_END;
req->analyse_exp = TICK_ETERNITY;
return 0;
@@ -4586,7 +4584,7 @@
if (unlikely((conn = si_alloc_conn(&s->si[1])) == NULL)) {
txn->req.msg_state = HTTP_MSG_ERROR;
txn->status = 500;
- req->analysers = 0;
+ req->analysers &= AN_FLT_END;
http_reply_and_close(s, txn->status, http_error_message(s, HTTP_ERR_500));
if (!(s->flags & SF_ERR_MASK))
@@ -4819,6 +4817,7 @@
}
if (msg->flags & HTTP_MSGF_XFER_LEN) {
+ req->analysers &= ~AN_FLT_XFER_DATA;
req->analysers |= AN_REQ_HTTP_XFER_BODY;
#ifdef TCP_QUICKACK
/* We expect some data from the client. Unless we know for sure
@@ -4861,7 +4860,7 @@
txn->req.msg_state = HTTP_MSG_ERROR;
txn->status = 400;
- req->analysers = 0;
+ req->analysers &= AN_FLT_END;
http_reply_and_close(s, txn->status, http_error_message(s, HTTP_ERR_400));
sess->fe->fe_counters.failed_req++;
@@ -4905,7 +4904,7 @@
if (!(req->flags & CF_READ_ERROR))
http_reply_and_close(s, txn->status, http_error_message(s, HTTP_ERR_500));
- req->analysers = 0;
+ req->analysers &= AN_FLT_END;
req->analyse_exp = TICK_ETERNITY;
if (!(s->flags & SF_ERR_MASK))
@@ -4994,6 +4993,9 @@
stream_inc_http_err_ctr(s);
goto return_bad_req;
}
+ msg->next += msg->sol;
+ msg->sol = 0;
+ msg->msg_state = msg->chunk_len ? HTTP_MSG_DATA : HTTP_MSG_TRAILERS;
}
/* Now we're in HTTP_MSG_DATA or HTTP_MSG_TRAILERS state.
@@ -5057,7 +5059,7 @@
s->flags |= SF_FINST_R;
return_err_msg:
- req->analysers = 0;
+ req->analysers &= AN_FLT_END;
sess->fe->fe_counters.failed_req++;
if (sess->listener->counters)
sess->listener->counters->failed_req++;
@@ -5086,7 +5088,7 @@
/* The request was already skipped, let's restore it */
b_rew(chn->buf, old_o);
txn->req.next += old_o;
- txn->req.sov += old_o;
+ txn->req.sov += old_o;
}
old_i = chn->buf->i;
@@ -5308,8 +5310,14 @@
si_idle_conn(&s->si[1], &srv->idle_conns);
}
- s->req.analysers = strm_li(s) ? strm_li(s)->analysers : 0;
- s->res.analysers = 0;
+ if (LIST_ISEMPTY(&s->strm_flt.filters)) {
+ s->req.analysers = strm_li(s) ? strm_li(s)->analysers : 0;
+ s->res.analysers = 0;
+ }
+ else {
+ s->req.analysers &= AN_FLT_END;
+ s->res.analysers &= AN_FLT_END;
+ }
}
@@ -5622,10 +5630,10 @@
txn->rsp.msg_state == HTTP_MSG_TUNNEL ||
(txn->req.msg_state == HTTP_MSG_CLOSED &&
txn->rsp.msg_state == HTTP_MSG_CLOSED)) {
- s->req.analysers = 0;
+ s->req.analysers &= AN_FLT_END;
channel_auto_close(&s->req);
channel_auto_read(&s->req);
- s->res.analysers = 0;
+ s->res.analysers &= AN_FLT_END;
channel_auto_close(&s->res);
channel_auto_read(&s->res);
}
@@ -5633,10 +5641,10 @@
(txn->rsp.msg_state == HTTP_MSG_CLOSED || (s->res.flags & CF_SHUTW))) ||
txn->rsp.msg_state == HTTP_MSG_ERROR ||
txn->req.msg_state == HTTP_MSG_ERROR) {
- s->res.analysers = 0;
+ s->res.analysers &= AN_FLT_END;
channel_auto_close(&s->res);
channel_auto_read(&s->res);
- s->req.analysers = 0;
+ s->req.analysers &= AN_FLT_END;
channel_abort(&s->req);
channel_auto_close(&s->req);
channel_auto_read(&s->req);
@@ -5673,6 +5681,7 @@
struct session *sess = s->sess;
struct http_txn *txn = s->txn;
struct http_msg *msg = &s->txn->req;
+ int ret, ret2;
if (unlikely(msg->msg_state < HTTP_MSG_BODY))
return 0;
@@ -5693,15 +5702,24 @@
* an "Expect: 100-continue" header.
*/
- if (msg->sov > 0) {
+ if (msg->msg_state == HTTP_MSG_BODY) {
/* we have msg->sov which points to the first byte of message
* body, and req->buf.p still points to the beginning of the
* message. We forward the headers now, as we don't need them
* anymore, and we want to flush them.
*/
- b_adv(req->buf, msg->sov);
- msg->next -= msg->sov;
- msg->sov = 0;
+ ret = flt_http_headers(s, msg);
+ if (ret < 0)
+ goto return_bad_req;
+ if (!ret)
+ return 0;
+
+ ret = flt_http_forward_data(s, msg, msg->sov);
+ if (ret < 0)
+ goto return_bad_req;
+ b_adv(req->buf, ret);
+ msg->next -= ret;
+ msg->sov -= ret;
/* The previous analysers guarantee that the state is somewhere
* between MSG_BODY and the first MSG_DATA. So msg->sol and
@@ -5713,6 +5731,12 @@
else
msg->msg_state = HTTP_MSG_DATA;
}
+
+ /* TODO/filters: when http-buffer-request option is set or if a
+ * rule on url_param exists, the first chunk size could be
+ * already parsed. In that case, msg->next is after the chunk
+ * size (including the CRLF after the size). So this case should
+ * be handled to */
}
/* Some post-connect processing might want us to refrain from starting to
@@ -5741,81 +5765,134 @@
if (msg->msg_state == HTTP_MSG_DATA) {
/* must still forward */
/* we may have some pending data starting at req->buf->p */
- if (msg->chunk_len > req->buf->i - msg->next) {
- req->flags |= CF_WAKE_WRITE;
+ ret = flt_http_data(s, msg);
+ if (ret < 0)
+ goto aborted_xfer;
+ msg->next += ret;
+ msg->chunk_len -= ret;
+
+ if (msg->chunk_len) {
+ /* input empty or output full */
+ if (req->buf->i > msg->next)
+ req->flags |= CF_WAKE_WRITE;
goto missing_data;
}
- msg->next += msg->chunk_len;
- msg->chunk_len = 0;
/* nothing left to forward */
if (msg->flags & HTTP_MSGF_TE_CHNK)
msg->msg_state = HTTP_MSG_CHUNK_CRLF;
else
- msg->msg_state = HTTP_MSG_DONE;
+ msg->msg_state = HTTP_MSG_ENDING;
}
else if (msg->msg_state == HTTP_MSG_CHUNK_SIZE) {
/* read the chunk size and assign it to ->chunk_len, then
* set ->next to point to the body and switch to DATA or
* TRAILERS state.
*/
- int ret = http_parse_chunk_size(msg);
-
- if (ret == 0)
- goto missing_data;
- else if (ret < 0) {
- stream_inc_http_err_ctr(s);
- if (msg->err_pos >= 0)
- http_capture_bad_message(&sess->fe->invalid_req, s, msg, HTTP_MSG_CHUNK_SIZE, s->be);
- goto return_bad_req;
+ if (!msg->sol) {
+ ret = http_parse_chunk_size(msg);
+ if (ret == 0)
+ goto missing_data;
+ else if (ret < 0) {
+ stream_inc_http_err_ctr(s);
+ if (msg->err_pos >= 0)
+ http_capture_bad_message(&sess->fe->invalid_req, s, msg, HTTP_MSG_CHUNK_SIZE, s->be);
+ goto return_bad_req;
+ }
}
+ ret = (msg->chunk_len
+ ? flt_http_start_chunk(s, msg)
+ : flt_http_last_chunk(s, msg));
+ if (ret < 0)
+ goto return_bad_req;
+ if (!ret)
+ goto missing_data;
+ msg->next += msg->sol;
+ msg->sol = 0;
+ msg->msg_state = msg->chunk_len ? HTTP_MSG_DATA : HTTP_MSG_TRAILERS;
/* otherwise we're in HTTP_MSG_DATA or HTTP_MSG_TRAILERS state */
}
else if (msg->msg_state == HTTP_MSG_CHUNK_CRLF) {
/* we want the CRLF after the data */
- int ret = http_skip_chunk_crlf(msg);
-
- if (ret == 0)
- goto missing_data;
- else if (ret < 0) {
- stream_inc_http_err_ctr(s);
- if (msg->err_pos >= 0)
- http_capture_bad_message(&sess->fe->invalid_req, s, msg, HTTP_MSG_CHUNK_CRLF, s->be);
- goto return_bad_req;
+ if (!msg->sol) {
+ ret = http_skip_chunk_crlf(msg);
+ if (ret == 0)
+ goto missing_data;
+ else if (ret < 0) {
+ stream_inc_http_err_ctr(s);
+ if (msg->err_pos >= 0)
+ http_capture_bad_message(&sess->fe->invalid_req, s, msg, HTTP_MSG_CHUNK_CRLF, s->be);
+ goto return_bad_req;
+ }
}
+ ret = flt_http_end_chunk(s, msg);
+ if (ret < 0)
+ goto return_bad_req;
+ if (!ret)
+ goto missing_data;
+ msg->next += msg->sol;
+ msg->sol = 0;
+ msg->msg_state = HTTP_MSG_CHUNK_SIZE;
/* we're in MSG_CHUNK_SIZE now */
}
else if (msg->msg_state == HTTP_MSG_TRAILERS) {
- int ret = http_forward_trailers(msg);
-
- if (ret == 0)
- goto missing_data;
- else if (ret < 0) {
- stream_inc_http_err_ctr(s);
- if (msg->err_pos >= 0)
- http_capture_bad_message(&sess->fe->invalid_req, s, msg, HTTP_MSG_TRAILERS, s->be);
- goto return_bad_req;
+ ret = 1;
+ if (!msg->sol) {
+ ret = http_forward_trailers(msg);
+ if (ret < 0) {
+ stream_inc_http_err_ctr(s);
+ if (msg->err_pos >= 0)
+ http_capture_bad_message(&sess->fe->invalid_req, s, msg, HTTP_MSG_TRAILERS, s->be);
+ goto return_bad_req;
+ }
}
- /* we're in HTTP_MSG_DONE now */
+ ret2 = flt_http_chunk_trailers(s, msg);
+ if (ret2 < 0)
+ goto return_bad_req;
+ if (!ret2)
+ goto missing_data;
+ msg->next += msg->sol;
+ msg->sol = 0;
+ if (!ret)
+ goto missing_data;
+ msg->msg_state = HTTP_MSG_ENDING;
}
- else {
- int old_state = msg->msg_state;
-
- /* other states, DONE...TUNNEL */
-
+ else if (msg->msg_state == HTTP_MSG_ENDING) {
+ /* we don't want to forward closes on DONE except in
+ * tunnel mode.
+ */
+ if ((txn->flags & TX_CON_WANT_MSK) != TX_CON_WANT_TUN)
+ channel_dont_close(req);
/* we may have some pending data starting at req->buf->p
* such as last chunk of data or trailers.
*/
- b_adv(req->buf, msg->next);
- if (unlikely(!(s->req.flags & CF_WROTE_DATA)))
- msg->sov -= msg->next;
- msg->next = 0;
+ ret = flt_http_forward_data(s, msg, msg->next);
+ if (ret < 0)
+ goto return_bad_req;
+ b_adv(req->buf, ret);
+ msg->next -= ret;
+ if (unlikely(!(s->req.flags & CF_WROTE_DATA) || msg->sov > 0))
+ msg->sov -= ret;
+ if (msg->next)
+ goto skip_resync_states;
+
+ ret = flt_http_end(s, msg);
+ if (ret < 0)
+ goto return_bad_req;
+ if (!ret)
+ goto skip_resync_states;
+ msg->msg_state = HTTP_MSG_DONE;
+ }
+ else {
+ /* other states, DONE...TUNNEL */
/* we don't want to forward closes on DONE except in
* tunnel mode.
*/
if ((txn->flags & TX_CON_WANT_MSK) != TX_CON_WANT_TUN)
channel_dont_close(req);
+
+ ret = msg->msg_state;
if (http_resync_states(s)) {
/* some state changes occurred, maybe the analyser
* was disabled too.
@@ -5828,12 +5905,13 @@
goto aborted_xfer;
}
if (msg->err_pos >= 0)
- http_capture_bad_message(&sess->fe->invalid_req, s, msg, old_state, s->be);
+ http_capture_bad_message(&sess->fe->invalid_req, s, msg, ret, s->be);
goto return_bad_req;
}
return 1;
}
+ skip_resync_states:
/* If "option abortonclose" is set on the backend, we
* want to monitor the client's connection and forward
* any shutdown notification to the server, which will
@@ -5864,12 +5942,15 @@
missing_data:
/* we may have some pending data starting at req->buf->p */
- b_adv(req->buf, msg->next);
- if (unlikely(!(s->req.flags & CF_WROTE_DATA)))
- msg->sov -= msg->next + MIN(msg->chunk_len, req->buf->i);
-
- msg->next = 0;
- msg->chunk_len -= channel_forward(req, msg->chunk_len);
+ ret = flt_http_forward_data(s, msg, msg->next);
+ if (ret < 0)
+ goto return_bad_req;
+ b_adv(req->buf, ret);
+ msg->next -= ret;
+ if (unlikely(!(s->req.flags & CF_WROTE_DATA) || msg->sov > 0))
+ msg->sov -= ret;
+ if (LIST_ISEMPTY(&s->strm_flt.filters))
+ msg->chunk_len -= channel_forward(req, msg->chunk_len);
/* stop waiting for data if the input is closed before the end */
if (req->flags & CF_SHUTR) {
@@ -5919,10 +6000,6 @@
sess->listener->counters->failed_req++;
return_bad_req_stats_ok:
- /* we may have some pending data starting at req->buf->p */
- b_adv(req->buf, msg->next);
- msg->next = 0;
-
txn->req.msg_state = HTTP_MSG_ERROR;
if (txn->status) {
/* Note: we don't send any error if some data were already sent */
@@ -5931,8 +6008,8 @@
txn->status = 400;
http_reply_and_close(s, txn->status, http_error_message(s, HTTP_ERR_400));
}
- req->analysers = 0;
- s->res.analysers = 0; /* we're in data phase, we want to abort both directions */
+ req->analysers &= AN_FLT_END;
+ s->res.analysers &= AN_FLT_END; /* we're in data phase, we want to abort both directions */
if (!(s->flags & SF_ERR_MASK))
s->flags |= SF_ERR_PRXCOND;
@@ -5953,8 +6030,8 @@
txn->status = 502;
http_reply_and_close(s, txn->status, http_error_message(s, HTTP_ERR_502));
}
- req->analysers = 0;
- s->res.analysers = 0; /* we're in data phase, we want to abort both directions */
+ req->analysers &= AN_FLT_END;
+ s->res.analysers &= AN_FLT_END; /* we're in data phase, we want to abort both directions */
sess->fe->fe_counters.srv_aborts++;
s->be->be_counters.srv_aborts++;
@@ -6090,7 +6167,7 @@
}
abort_response:
channel_auto_close(rep);
- rep->analysers = 0;
+ rep->analysers &= AN_FLT_END;
txn->status = 502;
s->si[1].flags |= SI_FL_NOLINGER;
channel_truncate(rep);
@@ -6125,7 +6202,7 @@
}
channel_auto_close(rep);
- rep->analysers = 0;
+ rep->analysers &= AN_FLT_END;
txn->status = 502;
s->si[1].flags |= SI_FL_NOLINGER;
channel_truncate(rep);
@@ -6150,7 +6227,7 @@
}
channel_auto_close(rep);
- rep->analysers = 0;
+ rep->analysers &= AN_FLT_END;
txn->status = 504;
s->si[1].flags |= SI_FL_NOLINGER;
channel_truncate(rep);
@@ -6170,7 +6247,7 @@
if (objt_server(s->target))
objt_server(s->target)->counters.cli_aborts++;
- rep->analysers = 0;
+ rep->analysers &= AN_FLT_END;
channel_auto_close(rep);
txn->status = 400;
@@ -6200,7 +6277,7 @@
}
channel_auto_close(rep);
- rep->analysers = 0;
+ rep->analysers &= AN_FLT_END;
txn->status = 502;
s->si[1].flags |= SI_FL_NOLINGER;
channel_truncate(rep);
@@ -6221,7 +6298,7 @@
goto abort_keep_alive;
s->be->be_counters.failed_resp++;
- rep->analysers = 0;
+ rep->analysers &= AN_FLT_END;
channel_auto_close(rep);
if (!(s->flags & SF_ERR_MASK))
@@ -6324,6 +6401,7 @@
msg->msg_state = HTTP_MSG_RPBEFORE;
txn->status = 0;
s->logs.t_data = -1; /* was not a response yet */
+ flt_http_reset(s, msg);
goto next_one;
case 200:
@@ -6579,8 +6657,8 @@
* any other information so that the client retries.
*/
txn->status = 0;
- rep->analysers = 0;
- s->req.analysers = 0;
+ rep->analysers &= AN_FLT_END;
+ s->req.analysers &= AN_FLT_END;
channel_auto_close(rep);
s->logs.logwait = 0;
s->logs.level = 0;
@@ -6680,7 +6758,7 @@
}
s->be->be_counters.failed_resp++;
return_srv_prx_502:
- rep->analysers = 0;
+ rep->analysers &= AN_FLT_END;
txn->status = 502;
s->logs.t_data = -1; /* was not a valid response */
s->si[1].flags |= SI_FL_NOLINGER;
@@ -6899,8 +6977,11 @@
skip_header_mangling:
if ((msg->flags & HTTP_MSGF_XFER_LEN) ||
- (txn->flags & TX_CON_WANT_MSK) == TX_CON_WANT_TUN)
+ !LIST_ISEMPTY(&s->strm_flt.filters) ||
+ (txn->flags & TX_CON_WANT_MSK) == TX_CON_WANT_TUN) {
+ rep->analysers &= ~AN_FLT_XFER_DATA;
rep->analysers |= AN_RES_HTTP_XFER_BODY;
+ }
/* if the user wants to log as soon as possible, without counting
* bytes from the server, then this is the right moment. We have
@@ -6949,16 +7030,14 @@
struct session *sess = s->sess;
struct http_txn *txn = s->txn;
struct http_msg *msg = &s->txn->rsp;
- static struct buffer *tmpbuf = &buf_empty;
- int compressing = 0;
- int ret;
+ int ret, ret2;
if (unlikely(msg->msg_state < HTTP_MSG_BODY))
return 0;
if ((res->flags & (CF_READ_ERROR|CF_READ_TIMEOUT|CF_WRITE_ERROR|CF_WRITE_TIMEOUT)) ||
((res->flags & CF_SHUTW) && (res->to_forward || res->buf->o)) ||
- !s->req.analysers) {
+ !s->req.analysers) {
/* Output closed while we were sending data. We must abort and
* wake the other side up.
*/
@@ -6970,15 +7049,24 @@
/* in most states, we should abort in case of early close */
channel_auto_close(res);
- if (msg->sov > 0) {
+ if (msg->msg_state == HTTP_MSG_BODY) {
/* we have msg->sov which points to the first byte of message
* body, and res->buf.p still points to the beginning of the
* message. We forward the headers now, as we don't need them
* anymore, and we want to flush them.
*/
- b_adv(res->buf, msg->sov);
- msg->next -= msg->sov;
- msg->sov = 0;
+ ret = flt_http_headers(s, msg);
+ if (ret < 0)
+ goto return_bad_res;
+ if (!ret)
+ return 0;
+
+ ret = flt_http_forward_data(s, msg, msg->sov);
+ if (ret < 0)
+ goto return_bad_res;
+ b_adv(res->buf, ret);
+ msg->next -= ret;
+ msg->sov -= ret;
/* The previous analysers guarantee that the state is somewhere
* between MSG_BODY and the first MSG_DATA. So msg->sol and
@@ -6998,71 +7086,62 @@
goto missing_data;
}
- if (unlikely(s->comp_algo != NULL) && msg->msg_state < HTTP_MSG_TRAILERS) {
- /* We need a compression buffer in the DATA state to put the
- * output of compressed data, and in CRLF state to let the
- * TRAILERS state finish the job of removing the trailing CRLF.
- */
- if (unlikely(!tmpbuf->size)) {
- /* this is the first time we need the compression buffer */
- if (b_alloc(&tmpbuf) == NULL)
- goto aborted_xfer; /* no memory */
- }
-
- ret = http_compression_buffer_init(s, res->buf, tmpbuf);
- if (ret < 0) {
- res->flags |= CF_WAKE_WRITE;
- goto missing_data; /* not enough spaces in buffers */
- }
- compressing = 1;
- }
-
while (1) {
switch (msg->msg_state - HTTP_MSG_DATA) {
case HTTP_MSG_DATA - HTTP_MSG_DATA: /* must still forward */
/* we may have some pending data starting at res->buf->p */
- if (unlikely(s->comp_algo)) {
- ret = http_compression_buffer_add_data(s, res->buf, tmpbuf);
- if (ret < 0)
- goto aborted_xfer;
- if (msg->chunk_len) {
- /* input empty or output full */
- if (res->buf->i > msg->next)
- res->flags |= CF_WAKE_WRITE;
- goto missing_data;
- }
- }
- else {
- if (msg->chunk_len > res->buf->i - msg->next) {
- /* output full */
+ /* Neither content-length, nor transfer-encoding was
+ * found, so we must read the body until the server
+ * connection is closed. In that case, we eat data as
+ * they come. */
+ if (!(msg->flags & HTTP_MSGF_XFER_LEN))
+ msg->chunk_len = (res->buf->i - msg->next);
+ ret = flt_http_data(s, msg);
+ if (ret < 0)
+ goto aborted_xfer;
+ msg->next += ret;
+ msg->chunk_len -= ret;
+ if (msg->chunk_len) {
+ /* input empty or output full */
+ if (res->buf->i > msg->next)
res->flags |= CF_WAKE_WRITE;
- goto missing_data;
- }
- msg->next += msg->chunk_len;
- msg->chunk_len = 0;
+ goto missing_data;
}
/* nothing left to forward */
if (msg->flags & HTTP_MSGF_TE_CHNK) {
msg->msg_state = HTTP_MSG_CHUNK_CRLF;
+ } else if (!(msg->flags & HTTP_MSGF_XFER_LEN) &&
+ !(res->flags & CF_SHUTR)) {
+ /* The server still sending data */
+ goto missing_data;
} else {
- msg->msg_state = HTTP_MSG_DONE;
+ msg->msg_state = HTTP_MSG_ENDING;
break;
}
/* fall through for HTTP_MSG_CHUNK_CRLF */
case HTTP_MSG_CHUNK_CRLF - HTTP_MSG_DATA:
/* we want the CRLF after the data */
-
- ret = http_skip_chunk_crlf(msg);
- if (ret == 0)
- goto missing_data;
- else if (ret < 0) {
- if (msg->err_pos >= 0)
- http_capture_bad_message(&s->be->invalid_rep, s, msg, HTTP_MSG_CHUNK_CRLF, sess->fe);
- goto return_bad_res;
+ if (!msg->sol) {
+ ret = http_skip_chunk_crlf(msg);
+ if (ret == 0)
+ goto missing_data;
+ else if (ret < 0) {
+ if (msg->err_pos >= 0)
+ http_capture_bad_message(&s->be->invalid_rep, s, msg, HTTP_MSG_CHUNK_CRLF, sess->fe);
+ goto return_bad_res;
+ }
}
+ ret = flt_http_end_chunk(s, msg);
+ if (ret < 0)
+ goto return_bad_res;
+ if (!ret)
+ goto missing_data;
+ msg->next += msg->sol;
+ msg->sol = 0;
+ msg->msg_state = HTTP_MSG_CHUNK_SIZE;
/* we're in MSG_CHUNK_SIZE now, fall through */
case HTTP_MSG_CHUNK_SIZE - HTTP_MSG_DATA:
@@ -7070,55 +7149,88 @@
* set ->next to point to the body and switch to DATA or
* TRAILERS state.
*/
-
- ret = http_parse_chunk_size(msg);
- if (ret == 0)
- goto missing_data;
- else if (ret < 0) {
- if (msg->err_pos >= 0)
- http_capture_bad_message(&s->be->invalid_rep, s, msg, HTTP_MSG_CHUNK_SIZE, sess->fe);
- goto return_bad_res;
+ if (!msg->sol) {
+ ret = http_parse_chunk_size(msg);
+ if (ret == 0)
+ goto missing_data;
+ else if (ret < 0) {
+ if (msg->err_pos >= 0)
+ http_capture_bad_message(&s->be->invalid_rep, s, msg, HTTP_MSG_CHUNK_SIZE, sess->fe);
+ goto return_bad_res;
+ }
}
+ ret = (msg->chunk_len
+ ? flt_http_start_chunk(s, msg)
+ : flt_http_last_chunk(s, msg));
+ if (ret < 0)
+ goto return_bad_res;
+ if (!ret)
+ goto missing_data;
+ msg->next += msg->sol;
+ msg->sol = 0;
+ msg->msg_state = msg->chunk_len ? HTTP_MSG_DATA : HTTP_MSG_TRAILERS;
/* otherwise we're in HTTP_MSG_DATA or HTTP_MSG_TRAILERS state */
break;
case HTTP_MSG_TRAILERS - HTTP_MSG_DATA:
- if (unlikely(compressing)) {
- /* we need to flush output contents before syncing FSMs */
- http_compression_buffer_end(s, &res->buf, &tmpbuf, 1);
- compressing = 0;
+ ret = 1;
+ if (!msg->sol) {
+ ret = http_forward_trailers(msg);
+ if (ret < 0) {
+ if (msg->err_pos >= 0)
+ http_capture_bad_message(&s->be->invalid_rep, s, msg, HTTP_MSG_TRAILERS, sess->fe);
+ goto return_bad_res;
+ }
}
-
- ret = http_forward_trailers(msg);
- if (ret == 0)
- goto missing_data;
- else if (ret < 0) {
- if (msg->err_pos >= 0)
- http_capture_bad_message(&s->be->invalid_rep, s, msg, HTTP_MSG_TRAILERS, sess->fe);
+ ret2 = flt_http_chunk_trailers(s, msg);
+ if (ret2 < 0)
goto return_bad_res;
- }
- /* we're in HTTP_MSG_DONE now, fall through */
+ if (!ret2)
+ goto missing_data;
+ msg->next += msg->sol;
+ msg->sol = 0;
+ if (!ret)
+ goto missing_data;
+ msg->msg_state = HTTP_MSG_ENDING;
+ /* fall through */
- default:
- /* other states, DONE...TUNNEL */
- if (unlikely(compressing)) {
- /* we need to flush output contents before syncing FSMs */
- http_compression_buffer_end(s, &res->buf, &tmpbuf, 1);
- compressing = 0;
- }
+ case HTTP_MSG_ENDING - HTTP_MSG_DATA:
+ /* for keep-alive we don't want to forward closes on ENDING */
+ if ((txn->flags & TX_CON_WANT_MSK) == TX_CON_WANT_KAL ||
+ (txn->flags & TX_CON_WANT_MSK) == TX_CON_WANT_SCL)
+ channel_dont_close(res);
/* we may have some pending data starting at res->buf->p
* such as a last chunk of data or trailers.
*/
- b_adv(res->buf, msg->next);
- msg->next = 0;
+ ret = flt_http_forward_data(s, msg, msg->next);
+ if (ret < 0)
+ goto return_bad_res;
+ b_adv(res->buf, ret);
+ msg->next -= ret;
+ if (msg->sov > 0)
+ msg->sov -= ret;
- ret = msg->msg_state;
+ if (msg->next)
+ goto skip_resync_states;
+
+ ret = flt_http_end(s, msg);
+ if (ret < 0)
+ goto return_bad_res;
+ if (!ret)
+ goto skip_resync_states;
+ msg->msg_state = HTTP_MSG_DONE;
+ /* fall through */
+
+ default:
+ /* other states, DONE...TUNNEL */
+
/* for keep-alive we don't want to forward closes on DONE */
if ((txn->flags & TX_CON_WANT_MSK) == TX_CON_WANT_KAL ||
(txn->flags & TX_CON_WANT_MSK) == TX_CON_WANT_SCL)
channel_dont_close(res);
+ ret = msg->msg_state;
if (http_resync_states(s)) {
/* some state changes occurred, maybe the analyser
* was disabled too.
@@ -7136,22 +7248,25 @@
}
return 1;
}
+
+ skip_resync_states:
return 0;
}
}
missing_data:
/* we may have some pending data starting at res->buf->p */
- if (unlikely(compressing)) {
- http_compression_buffer_end(s, &res->buf, &tmpbuf, msg->msg_state >= HTTP_MSG_TRAILERS);
- compressing = 0;
- }
+ ret = flt_http_forward_data(s, msg, msg->next);
+ if (ret < 0)
+ goto return_bad_res;
+ b_adv(res->buf, ret);
+ msg->next -= ret;
+ if (msg->sov > 0)
+ msg->sov -= ret;
- if ((s->comp_algo == NULL || msg->msg_state >= HTTP_MSG_TRAILERS)) {
- b_adv(res->buf, msg->next);
- msg->next = 0;
+ if ((s->comp_algo == NULL || msg->msg_state >= HTTP_MSG_TRAILERS) &&
+ LIST_ISEMPTY(&s->strm_flt.filters))
msg->chunk_len -= channel_forward(res, msg->chunk_len);
- }
if (res->flags & CF_SHUTW)
goto aborted_xfer;
@@ -7184,7 +7299,7 @@
* Similarly, with keep-alive on the client side, we don't want to forward a
* close.
*/
- if ((msg->flags & HTTP_MSGF_TE_CHNK) || s->comp_algo ||
+ if ((msg->flags & HTTP_MSGF_TE_CHNK) || s->comp_algo || !msg->body_len ||
(txn->flags & TX_CON_WANT_MSK) == TX_CON_WANT_KAL ||
(txn->flags & TX_CON_WANT_MSK) == TX_CON_WANT_SCL)
channel_dont_close(res);
@@ -7209,22 +7324,11 @@
objt_server(s->target)->counters.failed_resp++;
return_bad_res_stats_ok:
- if (unlikely(compressing)) {
- http_compression_buffer_end(s, &res->buf, &tmpbuf, msg->msg_state >= HTTP_MSG_TRAILERS);
- compressing = 0;
- }
-
- /* we may have some pending data starting at res->buf->p */
- if (s->comp_algo == NULL) {
- b_adv(res->buf, msg->next);
- msg->next = 0;
- }
-
txn->rsp.msg_state = HTTP_MSG_ERROR;
/* don't send any error message as we're in the body */
http_reply_and_close(s, txn->status, NULL);
- res->analysers = 0;
- s->req.analysers = 0; /* we're in data phase, we want to abort both directions */
+ res->analysers &= AN_FLT_END;
+ s->req.analysers &= AN_FLT_END; /* we're in data phase, we want to abort both directions */
if (objt_server(s->target))
health_adjust(objt_server(s->target), HANA_STATUS_HTTP_HDRRSP);
@@ -7235,16 +7339,11 @@
return 0;
aborted_xfer:
- if (unlikely(compressing)) {
- http_compression_buffer_end(s, &res->buf, &tmpbuf, msg->msg_state >= HTTP_MSG_TRAILERS);
- compressing = 0;
- }
-
txn->rsp.msg_state = HTTP_MSG_ERROR;
/* don't send any error message as we're in the body */
http_reply_and_close(s, txn->status, NULL);
- res->analysers = 0;
- s->req.analysers = 0; /* we're in data phase, we want to abort both directions */
+ res->analysers &= AN_FLT_END;
+ s->req.analysers &= AN_FLT_END; /* we're in data phase, we want to abort both directions */
sess->fe->fe_counters.cli_aborts++;
s->be->be_counters.cli_aborts++;
diff --git a/src/proxy.c b/src/proxy.c
index 3939bee..2014c73 100644
--- a/src/proxy.c
+++ b/src/proxy.c
@@ -35,6 +35,7 @@
#include <proto/backend.h>
#include <proto/fd.h>
+#include <proto/filters.h>
#include <proto/hdr_idx.h>
#include <proto/listener.h>
#include <proto/log.h>
@@ -747,6 +748,7 @@
LIST_INIT(&p->conf.listeners);
LIST_INIT(&p->conf.args.list);
LIST_INIT(&p->tcpcheck_rules);
+ LIST_INIT(&p->filters);
/* Timeouts are defined as -1 */
proxy_reset_timeouts(p);
@@ -1128,6 +1130,8 @@
*/
int stream_set_backend(struct stream *s, struct proxy *be)
{
+ struct filter *filter;
+
if (s->flags & SF_BE_ASSIGNED)
return 1;
s->be = be;
@@ -1136,6 +1140,20 @@
be->be_counters.conn_max = be->beconn;
proxy_inc_be_ctr(be);
+ if (strm_fe(s) != be) {
+ list_for_each_entry(filter, &be->filters, list) {
+ struct filter *f = pool_alloc2(pool2_filter);
+ if (!f)
+ return 0; /* not enough memory */
+ memset(f, 0, sizeof(*f));
+ f->id = filter->id;
+ f->ops = filter->ops;
+ f->conf = filter->conf;
+ f->is_backend_filter = 1;
+ LIST_ADDQ(&s->strm_flt.filters, &f->list);
+ }
+ }
+
/* assign new parameters to the stream from the new backend */
s->si[1].flags &= ~SI_FL_INDEP_STR;
if (be->options2 & PR_O2_INDEPSTR)
@@ -1146,9 +1164,7 @@
* be more reliable to store the list of analysers that have been run,
* but what we do here is OK for now.
*/
- s->req.analysers |= be->be_req_ana;
- if (strm_li(s))
- s->req.analysers &= ~strm_li(s)->analysers;
+ s->req.analysers |= be->be_req_ana & (strm_li(s) ? ~strm_li(s)->analysers : 0);
/* If the target backend requires HTTP processing, we have to allocate
* the HTTP transaction and hdr_idx if we did not have one.
diff --git a/src/session.c b/src/session.c
index 44fccbf..fdb2404 100644
--- a/src/session.c
+++ b/src/session.c
@@ -266,8 +266,9 @@
if (!strm)
goto out_free_task;
- strm->target = sess->listener->default_target;
- strm->req.analysers = sess->listener->analysers;
+ strm->target = sess->listener->default_target;
+ strm->req.analysers |= sess->listener->analysers;
+
return 1;
out_free_task:
@@ -431,8 +432,8 @@
if (!strm)
goto fail;
- strm->target = sess->listener->default_target;
- strm->req.analysers = sess->listener->analysers;
+ strm->target = sess->listener->default_target;
+ strm->req.analysers |= sess->listener->analysers;
conn->flags &= ~CO_FL_INIT_DATA;
return 0;
diff --git a/src/stream.c b/src/stream.c
index fe48be4..a98ecb0 100644
--- a/src/stream.c
+++ b/src/stream.c
@@ -22,6 +22,7 @@
#include <types/applet.h>
#include <types/capture.h>
+#include <types/filters.h>
#include <types/global.h>
#include <proto/acl.h>
@@ -33,6 +34,7 @@
#include <proto/connection.h>
#include <proto/dumpstats.h>
#include <proto/fd.h>
+#include <proto/filters.h>
#include <proto/freq_ctr.h>
#include <proto/frontend.h>
#include <proto/hdr_idx.h>
@@ -74,6 +76,7 @@
struct stream *stream_new(struct session *sess, struct task *t, enum obj_type *origin)
{
struct stream *s;
+ struct filter *filter, *back;
struct connection *conn = objt_conn(origin);
struct appctx *appctx = objt_appctx(origin);
@@ -214,6 +217,21 @@
HLUA_INIT(&s->hlua);
+ LIST_INIT(&s->strm_flt.filters);
+ memset(s->strm_flt.current, 0, sizeof(s->strm_flt.current));
+ list_for_each_entry(filter, &sess->fe->filters, list) {
+ struct filter *f = pool_alloc2(pool2_filter);
+ if (!f)
+ goto out_fail_accept;
+ memset(f, 0, sizeof(*f));
+ f->id = filter->id;
+ f->ops = filter->ops;
+ f->conf = filter->conf;
+ LIST_ADDQ(&s->strm_flt.filters, &f->list);
+ }
+ if (flt_stream_start(s) < 0)
+ goto out_fail_accept;
+
/* finish initialization of the accepted file descriptor */
if (conn)
conn_data_want_recv(conn);
@@ -232,6 +250,10 @@
/* Error unrolling */
out_fail_accept:
+ list_for_each_entry_safe(filter, back, &s->strm_flt.filters, list) {
+ LIST_DEL(&filter->list);
+ pool_free2(pool2_filter, filter);
+ }
LIST_DEL(&s->list);
pool_free2(pool2_stream, s);
return NULL;
@@ -246,6 +268,7 @@
struct proxy *fe = sess->fe;
struct bref *bref, *back;
struct connection *cli_conn = objt_conn(sess->origin);
+ struct filter *filter, *fback;
int i;
if (s->pend_pos)
@@ -306,6 +329,12 @@
s->txn = NULL;
}
+ flt_stream_stop(s);
+ list_for_each_entry_safe(filter, fback, &s->strm_flt.filters, list) {
+ LIST_DEL(&filter->list);
+ pool_free2(pool2_filter, filter);
+ }
+
if (fe) {
pool_free2(fe->rsp_cap_pool, s->res_cap);
pool_free2(fe->req_cap_pool, s->req_cap);
@@ -1166,6 +1195,7 @@
if (fe == s->be) {
s->req.analysers &= ~AN_REQ_INSPECT_BE;
s->req.analysers &= ~AN_REQ_HTTP_PROCESS_BE;
+ s->req.analysers &= ~AN_FLT_START_BE;
}
/* as soon as we know the backend, we must check if we have a matching forced or ignored
@@ -1206,7 +1236,7 @@
if (s->txn)
s->txn->status = 500;
- s->req.analysers = 0;
+ s->req.analysers &= AN_FLT_END;
s->req.analyse_exp = TICK_ETERNITY;
return 0;
}
@@ -1749,84 +1779,131 @@
ana_list = ana_back = req->analysers;
while (ana_list && max_loops--) {
/* Warning! ensure that analysers are always placed in ascending order! */
+ if (ana_list & AN_FLT_START_FE) {
+ if (!flt_start_analyze(s, req, AN_FLT_START_FE))
+ break;
+ UPDATE_ANALYSERS(req->analysers, ana_list, ana_back, AN_FLT_START_FE);
+ }
if (ana_list & AN_REQ_INSPECT_FE) {
+ if (!flt_analyze(s, req, AN_REQ_INSPECT_FE))
+ break;
if (!tcp_inspect_request(s, req, AN_REQ_INSPECT_FE))
break;
UPDATE_ANALYSERS(req->analysers, ana_list, ana_back, AN_REQ_INSPECT_FE);
}
if (ana_list & AN_REQ_WAIT_HTTP) {
+ if (!flt_analyze(s, req, AN_REQ_WAIT_HTTP))
+ break;
if (!http_wait_for_request(s, req, AN_REQ_WAIT_HTTP))
break;
UPDATE_ANALYSERS(req->analysers, ana_list, ana_back, AN_REQ_WAIT_HTTP);
}
if (ana_list & AN_REQ_HTTP_BODY) {
+ if (!flt_analyze(s, req, AN_REQ_HTTP_BODY))
+ break;
if (!http_wait_for_request_body(s, req, AN_REQ_HTTP_BODY))
break;
UPDATE_ANALYSERS(req->analysers, ana_list, ana_back, AN_REQ_HTTP_BODY);
}
if (ana_list & AN_REQ_HTTP_PROCESS_FE) {
+ if (!flt_analyze(s, req, AN_REQ_HTTP_PROCESS_FE))
+ break;
if (!http_process_req_common(s, req, AN_REQ_HTTP_PROCESS_FE, sess->fe))
break;
UPDATE_ANALYSERS(req->analysers, ana_list, ana_back, AN_REQ_HTTP_PROCESS_FE);
}
if (ana_list & AN_REQ_SWITCHING_RULES) {
+ if (!flt_analyze(s, req, AN_REQ_SWITCHING_RULES))
+ break;
if (!process_switching_rules(s, req, AN_REQ_SWITCHING_RULES))
break;
UPDATE_ANALYSERS(req->analysers, ana_list, ana_back, AN_REQ_SWITCHING_RULES);
}
if (ana_list & AN_REQ_INSPECT_BE) {
+ if (!flt_analyze(s, req, AN_REQ_INSPECT_BE))
+ break;
if (!tcp_inspect_request(s, req, AN_REQ_INSPECT_BE))
break;
UPDATE_ANALYSERS(req->analysers, ana_list, ana_back, AN_REQ_INSPECT_BE);
}
if (ana_list & AN_REQ_HTTP_PROCESS_BE) {
+ if (!flt_analyze(s, req, AN_REQ_HTTP_PROCESS_BE))
+ break;
if (!http_process_req_common(s, req, AN_REQ_HTTP_PROCESS_BE, s->be))
break;
UPDATE_ANALYSERS(req->analysers, ana_list, ana_back, AN_REQ_HTTP_PROCESS_BE);
}
if (ana_list & AN_REQ_HTTP_TARPIT) {
+ if (!flt_analyze(s, req, AN_REQ_HTTP_TARPIT))
+ break;
if (!http_process_tarpit(s, req, AN_REQ_HTTP_TARPIT))
break;
UPDATE_ANALYSERS(req->analysers, ana_list, ana_back, AN_REQ_HTTP_TARPIT);
}
if (ana_list & AN_REQ_SRV_RULES) {
+ if (!flt_analyze(s, req, AN_REQ_SRV_RULES))
+ break;
if (!process_server_rules(s, req, AN_REQ_SRV_RULES))
break;
UPDATE_ANALYSERS(req->analysers, ana_list, ana_back, AN_REQ_SRV_RULES);
}
if (ana_list & AN_REQ_HTTP_INNER) {
+ if (!flt_analyze(s, req, AN_REQ_HTTP_INNER))
+ break;
if (!http_process_request(s, req, AN_REQ_HTTP_INNER))
break;
UPDATE_ANALYSERS(req->analysers, ana_list, ana_back, AN_REQ_HTTP_INNER);
}
if (ana_list & AN_REQ_PRST_RDP_COOKIE) {
+ if (!flt_analyze(s, req, AN_REQ_PRST_RDP_COOKIE))
+ break;
if (!tcp_persist_rdp_cookie(s, req, AN_REQ_PRST_RDP_COOKIE))
break;
UPDATE_ANALYSERS(req->analysers, ana_list, ana_back, AN_REQ_PRST_RDP_COOKIE);
}
if (ana_list & AN_REQ_STICKING_RULES) {
+ if (!flt_analyze(s, req, AN_REQ_STICKING_RULES))
+ break;
if (!process_sticking_rules(s, req, AN_REQ_STICKING_RULES))
break;
UPDATE_ANALYSERS(req->analysers, ana_list, ana_back, AN_REQ_STICKING_RULES);
}
+ if (ana_list & AN_FLT_START_BE) {
+ if (!flt_start_analyze(s, req, AN_FLT_START_BE))
+ break;
+ UPDATE_ANALYSERS(req->analysers, ana_list, ana_back, AN_FLT_START_BE);
+ }
+
+ if (ana_list & AN_FLT_XFER_DATA) {
+ if (!flt_xfer_data(s, req, AN_FLT_XFER_DATA))
+ break;
+ UPDATE_ANALYSERS(req->analysers, ana_list, ana_back, AN_FLT_XFER_DATA);
+ }
+
if (ana_list & AN_REQ_HTTP_XFER_BODY) {
if (!http_request_forward_body(s, req, AN_REQ_HTTP_XFER_BODY))
break;
UPDATE_ANALYSERS(req->analysers, ana_list, ana_back, AN_REQ_HTTP_XFER_BODY);
}
+
+ if (ana_list & AN_FLT_END) {
+ if (!flt_end_analyze(s, req, AN_FLT_END))
+ break;
+ UPDATE_ANALYSERS(req->analysers, ana_list, ana_back, AN_FLT_END);
+ }
break;
}
}
@@ -1896,36 +1973,67 @@
ana_list = ana_back = res->analysers;
while (ana_list && max_loops--) {
/* Warning! ensure that analysers are always placed in ascending order! */
+ if (ana_list & AN_FLT_START_FE) {
+ if (!flt_start_analyze(s, res, AN_FLT_START_FE))
+ break;
+ UPDATE_ANALYSERS(res->analysers, ana_list, ana_back, AN_FLT_START_FE);
+ }
if (ana_list & AN_RES_INSPECT) {
+ if (!flt_analyze(s, res, AN_RES_INSPECT))
+ break;
if (!tcp_inspect_response(s, res, AN_RES_INSPECT))
break;
UPDATE_ANALYSERS(res->analysers, ana_list, ana_back, AN_RES_INSPECT);
}
if (ana_list & AN_RES_WAIT_HTTP) {
+ if (!flt_analyze(s, res, AN_RES_WAIT_HTTP))
+ break;
if (!http_wait_for_response(s, res, AN_RES_WAIT_HTTP))
break;
UPDATE_ANALYSERS(res->analysers, ana_list, ana_back, AN_RES_WAIT_HTTP);
}
if (ana_list & AN_RES_STORE_RULES) {
+ if (!flt_analyze(s, res, AN_RES_STORE_RULES))
+ break;
if (!process_store_rules(s, res, AN_RES_STORE_RULES))
break;
UPDATE_ANALYSERS(res->analysers, ana_list, ana_back, AN_RES_STORE_RULES);
}
if (ana_list & AN_RES_HTTP_PROCESS_BE) {
+ if (!flt_analyze(s, res, AN_RES_HTTP_PROCESS_BE))
+ break;
if (!http_process_res_common(s, res, AN_RES_HTTP_PROCESS_BE, s->be))
break;
UPDATE_ANALYSERS(res->analysers, ana_list, ana_back, AN_RES_HTTP_PROCESS_BE);
}
+ if (ana_list & AN_FLT_START_BE) {
+ if (!flt_start_analyze(s, res, AN_FLT_START_BE))
+ break;
+ UPDATE_ANALYSERS(res->analysers, ana_list, ana_back, AN_FLT_START_BE);
+ }
+
+ if (ana_list & AN_FLT_XFER_DATA) {
+ if (!flt_xfer_data(s, res, AN_FLT_XFER_DATA))
+ break;
+ UPDATE_ANALYSERS(res->analysers, ana_list, ana_back, AN_FLT_XFER_DATA);
+ }
+
if (ana_list & AN_RES_HTTP_XFER_BODY) {
if (!http_response_forward_body(s, res, AN_RES_HTTP_XFER_BODY))
break;
UPDATE_ANALYSERS(res->analysers, ana_list, ana_back, AN_RES_HTTP_XFER_BODY);
}
+
+ if (ana_list & AN_FLT_END) {
+ if (!flt_end_analyze(s, res, AN_FLT_END))
+ break;
+ UPDATE_ANALYSERS(res->analysers, ana_list, ana_back, AN_FLT_END);
+ }
break;
}
}