RepoMirrors/haproxy
1
0
Fork 0
mirror of http://git.haproxy.org/git/haproxy.git/ synced 2025-04-01 22:48:25 +00:00
Code Issues Packages Projects Releases Wiki Activity

MAJOR: filters: Add filters support

Browse Source 
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.
This commit is contained in:
Christopher Faulet 2015-04-30 11:48:27 +02:00 committed by Willy Tarreau
parent 0d6b7a490e
commit d7c9196ae5
16 changed files with 1842 additions and 257 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
Makefile
View File

@ -750,7 +750,7 @@ OBJS = src/haproxy.o src/base64.o src/protocol.o \
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 \

138
include/proto/filters.h Normal file
View File

@ -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 */

2
include/proto/proto_http.h
View File

@ -126,6 +126,7 @@ struct http_txn *http_alloc_txn(struct stream *s);
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 @@ static inline const char *http_msg_state_str(int msg_state)
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";

7
include/types/channel.h
View File

@ -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)

233
include/types/filters.h Normal file
View File

@ -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:
*/

12
include/types/proto_http.h
View File

@ -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 @@ enum ht_state {
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 @@ enum ht_state {
* 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 */

3
include/types/proxy.h
View File

@ -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 @@ struct proxy {
* this backend. If not specified or void, then the backend
* name is used
*/
struct list filters;
};
struct switching_rule {

7
include/types/stream.h
View File

@ -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 stream {
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 */

17
src/cfgparse.c
View File

@ -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 @@ out_uri_auth_compat:
}
}
/* 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 @@ out_uri_auth_compat:
/* 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 @@ out_uri_auth_compat:
*/
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;
}
}
}

7
src/compression.c
View File

@ -218,11 +218,6 @@ int http_compression_buffer_add_data(struct stream *s, struct buffer *in, struct
/* 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 @@ int http_compression_buffer_end(struct stream *s, struct buffer **in, struct buf
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;
}

946
src/filters.c Normal file
View File

@ -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:
*/

15
src/haproxy.c
View File

@ -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 @@ void init(int argc, char **argv)
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 @@ void init(int argc, char **argv)
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 @@ void deinit(void)
free(bind_conf);
}
flt_deinit(p);
free(p->desc);
free(p->fwdfor_hdr_name);
@ -1550,7 +1564,6 @@ void deinit(void)
pool_destroy2(pool2_sig_handlers);
pool_destroy2(pool2_hdr_idx);
pool_destroy2(pool2_http_txn);
deinit_pollers();
} /* end deinit() */

569
src/proto_http.c
View File

File diff suppressed because it is too large Load Diff

22
src/proxy.c
View File

@ -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 @@ void init_new_proxy(struct proxy *p)
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 @@ void resume_proxies(void)
*/
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 @@ int stream_set_backend(struct stream *s, struct proxy *be)
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 @@ int stream_set_backend(struct stream *s, struct proxy *be)
* 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.

9
src/session.c
View File

@ -266,8 +266,9 @@ int session_accept_fd(struct listener *l, int cfd, struct sockaddr_storage *addr
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 @@ static int conn_complete_session(struct connection *conn)
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;

110
src/stream.c
View File

@ -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 @@ static struct list service_keywords = LIST_HEAD_INIT(service_keywords);
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 @@ struct stream *stream_new(struct session *sess, struct task *t, enum obj_type *o
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 @@ struct stream *stream_new(struct session *sess, struct task *t, enum obj_type *o
/* 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 @@ static void stream_free(struct stream *s)
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 @@ static void stream_free(struct stream *s)
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 @@ static int process_switching_rules(struct stream *s, struct channel *req, int an
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 @@ static int process_switching_rules(struct stream *s, struct channel *req, int an
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 @@ struct task *process_stream(struct task *t)
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 @@ struct task *process_stream(struct task *t)
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;
}
}