diff options
Diffstat (limited to 'src/filters.c')
| -rw-r--r-- | src/filters.c | 1125 |
1 files changed, 1125 insertions, 0 deletions
diff --git a/src/filters.c b/src/filters.c new file mode 100644 index 0000000..e55adee --- /dev/null +++ b/src/filters.c @@ -0,0 +1,1125 @@ +/* + * 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 <haproxy/api.h> +#include <haproxy/buf-t.h> +#include <haproxy/cfgparse.h> +#include <haproxy/compression.h> +#include <haproxy/errors.h> +#include <haproxy/filters.h> +#include <haproxy/flt_http_comp.h> +#include <haproxy/http_ana.h> +#include <haproxy/http_htx.h> +#include <haproxy/htx.h> +#include <haproxy/namespace.h> +#include <haproxy/proxy.h> +#include <haproxy/stream.h> +#include <haproxy/tools.h> +#include <haproxy/trace.h> + + +#define TRACE_SOURCE &trace_strm + +/* Pool used to allocate filters */ +DECLARE_STATIC_POOL(pool_head_filter, "filter", sizeof(struct filter)); + +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 example: + * + * 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_flt(strm)->current[CHN_IDX(chn)]) { \ + filter = strm_flt(strm)->current[CHN_IDX(chn)]; \ + strm_flt(strm)->current[CHN_IDX(chn)] = NULL; \ + goto resume_execution; \ + } \ + \ + list_for_each_entry(filter, &strm_flt(s)->filters, list) { \ + resume_execution: + +#define RESUME_FILTER_END \ + } \ + } while(0) + +#define BREAK_EXECUTION(strm, chn, label) \ + do { \ + strm_flt(strm)->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_APPEND(&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. + * If <out> is NULL, the output is emitted using a more compact format on stdout. + */ +void +flt_dump_kws(char **out) +{ + struct flt_kw_list *kwl; + const struct flt_kw *kwp, *kw; + const char *scope = NULL; + int index; + + if (out) + *out = NULL; + + for (kw = kwp = NULL;; kwp = kw) { + 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]) + && strordered(kwp ? kwp->kw : NULL, + kwl->kw[index].kw, + kw != kwp ? kw->kw : NULL)) { + kw = &kwl->kw[index]; + scope = kwl->scope; + } + } + } + + if (kw == kwp) + break; + + if (out) + memprintf(out, "%s[%4s] %s%s\n", *out ? *out : "", + scope, + kw->kw, + kw->parse ? "" : " (not supported)"); + else + printf("%s [%s]\n", + kw->kw, scope); + } +} + +/* + * Lists the known filters on <out> + */ +void +list_filters(FILE *out) +{ + char *filters, *p, *f; + + fprintf(out, "Available filters :\n"); + flt_dump_kws(&filters); + for (p = filters; (f = strtok_r(p,"\n",&p));) + fprintf(out, "\t%s\n", f); + free(filters); +} + +/* + * Parses the "filter" keyword. All keywords must be handled by filters + * themselves + */ +static int +parse_filter(char **args, int section_type, struct proxy *curpx, + const struct proxy *defpx, const char *file, int line, char **err) +{ + struct flt_conf *fconf = NULL; + + /* Filter cannot be defined on a default proxy */ + if (curpx == defpx) { + memprintf(err, "parsing [%s:%d] : %s is not allowed in a 'default' section.", + file, line, args[0]); + return -1; + } + if (strcmp(args[0], "filter") == 0) { + 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; + } + fconf = calloc(1, sizeof(*fconf)); + if (!fconf) { + memprintf(err, "'%s' : out of memory", args[0]); + goto error; + } + + 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, fconf, err, kw->private) != 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; + } + if (fconf->ops == NULL) { + memprintf(err, "'%s %s' : no callbacks defined.", + args[0], args[1]); + goto error; + } + + LIST_APPEND(&curpx->filter_configs, &fconf->list); + } + return 0; + + error: + free(fconf); + 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. + */ +static int +flt_init(struct proxy *proxy) +{ + struct flt_conf *fconf; + + list_for_each_entry(fconf, &proxy->filter_configs, list) { + if (fconf->ops->init && fconf->ops->init(proxy, fconf) < 0) + return ERR_ALERT|ERR_FATAL; + } + return 0; +} + +/* + * Calls 'init_per_thread' callback for all filters attached to a proxy for each + * threads. This happens after the thread creation. Filters can finish to fill + * their config. Returns (ERR_ALERT|ERR_FATAL) if an error occurs, 0 otherwise. + */ +static int +flt_init_per_thread(struct proxy *proxy) +{ + struct flt_conf *fconf; + + list_for_each_entry(fconf, &proxy->filter_configs, list) { + if (fconf->ops->init_per_thread && fconf->ops->init_per_thread(proxy, fconf) < 0) + return ERR_ALERT|ERR_FATAL; + } + return 0; +} + +/* Calls flt_init() for all proxies, see above */ +static int +flt_init_all() +{ + struct proxy *px; + int err_code = ERR_NONE; + + for (px = proxies_list; px; px = px->next) { + if (px->flags & (PR_FL_DISABLED|PR_FL_STOPPED)) + continue; + + err_code |= flt_init(px); + if (err_code & (ERR_ABORT|ERR_FATAL)) { + ha_alert("Failed to initialize filters for proxy '%s'.\n", + px->id); + return err_code; + } + } + return 0; +} + +/* Calls flt_init_per_thread() for all proxies, see above. Be careful here, it + * returns 0 if an error occurred. This is the opposite of flt_init_all. */ +static int +flt_init_all_per_thread() +{ + struct proxy *px; + int err_code = 0; + + for (px = proxies_list; px; px = px->next) { + if (px->flags & (PR_FL_DISABLED|PR_FL_STOPPED)) + continue; + + err_code = flt_init_per_thread(px); + if (err_code & (ERR_ABORT|ERR_FATAL)) { + ha_alert("Failed to initialize filters for proxy '%s' for thread %u.\n", + px->id, tid); + return 0; + } + } + return 1; +} + +/* + * 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 flt_conf *fconf; + int err = 0; + + err += check_implicit_http_comp_flt(proxy); + list_for_each_entry(fconf, &proxy->filter_configs, list) { + if (fconf->ops->check) + err += fconf->ops->check(proxy, fconf); + } + 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 flt_conf *fconf, *back; + + list_for_each_entry_safe(fconf, back, &proxy->filter_configs, list) { + if (fconf->ops->deinit) + fconf->ops->deinit(proxy, fconf); + LIST_DELETE(&fconf->list); + free(fconf); + } +} + +/* + * Calls 'denit_per_thread' callback for all filters attached to a proxy for + * each threads. This happens before exiting a thread. + */ +void +flt_deinit_per_thread(struct proxy *proxy) +{ + struct flt_conf *fconf, *back; + + list_for_each_entry_safe(fconf, back, &proxy->filter_configs, list) { + if (fconf->ops->deinit_per_thread) + fconf->ops->deinit_per_thread(proxy, fconf); + } +} + + +/* Calls flt_deinit_per_thread() for all proxies, see above */ +static void +flt_deinit_all_per_thread() +{ + struct proxy *px; + + for (px = proxies_list; px; px = px->next) + flt_deinit_per_thread(px); +} + +/* Attaches a filter to a stream. Returns -1 if an error occurs, 0 otherwise. */ +static int +flt_stream_add_filter(struct stream *s, struct flt_conf *fconf, unsigned int flags) +{ + struct filter *f; + + if (IS_HTX_STRM(s) && !(fconf->flags & FLT_CFG_FL_HTX)) + return 0; + + f = pool_zalloc(pool_head_filter); + if (!f) /* not enough memory */ + return -1; + f->config = fconf; + f->flags |= flags; + + if (FLT_OPS(f)->attach) { + int ret = FLT_OPS(f)->attach(s, f); + if (ret <= 0) { + pool_free(pool_head_filter, f); + return ret; + } + } + + LIST_APPEND(&strm_flt(s)->filters, &f->list); + strm_flt(s)->flags |= STRM_FLT_FL_HAS_FILTERS; + return 0; +} + +/* + * Called when a stream is created. It attaches all frontend filters to the + * stream. Returns -1 if an error occurs, 0 otherwise. + */ +int +flt_stream_init(struct stream *s) +{ + struct flt_conf *fconf; + + memset(strm_flt(s), 0, sizeof(*strm_flt(s))); + LIST_INIT(&strm_flt(s)->filters); + list_for_each_entry(fconf, &strm_fe(s)->filter_configs, list) { + if (flt_stream_add_filter(s, fconf, 0) < 0) + return -1; + } + return 0; +} + +/* + * Called when a stream is closed or when analyze ends (For an HTTP stream, this + * happens after each request/response exchange). When analyze ends, backend + * filters are removed. When the stream is closed, all filters attached to the + * stream are removed. + */ +void +flt_stream_release(struct stream *s, int only_backend) +{ + struct filter *filter, *back; + + list_for_each_entry_safe(filter, back, &strm_flt(s)->filters, list) { + if (!only_backend || (filter->flags & FLT_FL_IS_BACKEND_FILTER)) { + if (FLT_OPS(filter)->detach) + FLT_OPS(filter)->detach(s, filter); + LIST_DELETE(&filter->list); + pool_free(pool_head_filter, filter); + } + } + if (LIST_ISEMPTY(&strm_flt(s)->filters)) + strm_flt(s)->flags &= ~STRM_FLT_FL_HAS_FILTERS; +} + +/* + * 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, &strm_flt(s)->filters, list) { + if (FLT_OPS(filter)->stream_start && FLT_OPS(filter)->stream_start(s, filter) < 0) + return -1; + } + if (strm_li(s) && (strm_li(s)->bind_conf->analysers & AN_REQ_FLT_START_FE)) { + s->req.flags |= CF_FLT_ANALYZE; + s->req.analysers |= AN_REQ_FLT_END; + } + 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, &strm_flt(s)->filters, list) { + if (FLT_OPS(filter)->stream_stop) + FLT_OPS(filter)->stream_stop(s, filter); + } +} + +/* + * Calls 'check_timeouts' for all filters attached to a stream. This happens when + * the stream is woken up because of expired timer. + */ +void +flt_stream_check_timeouts(struct stream *s) +{ + struct filter *filter; + + list_for_each_entry(filter, &strm_flt(s)->filters, list) { + if (FLT_OPS(filter)->check_timeouts) + FLT_OPS(filter)->check_timeouts(s, filter); + } +} + +/* + * Called when a backend is set for a stream. If the frontend and the backend + * are not the same, this function attaches all backend filters to the + * stream. Returns -1 if an error occurs, 0 otherwise. + */ +int +flt_set_stream_backend(struct stream *s, struct proxy *be) +{ + struct flt_conf *fconf; + struct filter *filter; + + if (strm_fe(s) == be) + goto end; + + list_for_each_entry(fconf, &be->filter_configs, list) { + if (flt_stream_add_filter(s, fconf, FLT_FL_IS_BACKEND_FILTER) < 0) + return -1; + } + + end: + list_for_each_entry(filter, &strm_flt(s)->filters, list) { + if (FLT_OPS(filter)->stream_set_backend && + FLT_OPS(filter)->stream_set_backend(s, filter, be) < 0) + return -1; + } + if (be->be_req_ana & AN_REQ_FLT_START_BE) { + s->req.flags |= CF_FLT_ANALYZE; + s->req.analysers |= AN_REQ_FLT_END; + } + if ((strm_fe(s)->fe_rsp_ana | be->be_rsp_ana) & (AN_RES_FLT_START_FE|AN_RES_FLT_START_BE)) { + s->res.flags |= CF_FLT_ANALYZE; + s->res.analysers |= AN_RES_FLT_END; + } + + return 0; +} + + +/* + * 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) +{ + unsigned long long *strm_off = &FLT_STRM_OFF(s, msg->chn); + unsigned int offset = 0; + int ret = 1; + + DBG_TRACE_ENTER(STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA|STRM_EV_FLT_ANA, s, s->txn, msg); + RESUME_FILTER_LOOP(s, msg->chn) { + unsigned long long flt_off = FLT_OFF(filter, msg->chn); + offset = flt_off - *strm_off; + + /* Call http_end for data filters only. But the filter offset is + * still valid for all filters + . */ + if (!IS_DATA_FILTER(filter, msg->chn)) + continue; + + if (FLT_OPS(filter)->http_end) { + DBG_TRACE_DEVEL(FLT_ID(filter), STRM_EV_HTTP_ANA|STRM_EV_FLT_ANA, s); + ret = FLT_OPS(filter)->http_end(s, filter, msg); + if (ret <= 0) + BREAK_EXECUTION(s, msg->chn, end); + } + } RESUME_FILTER_END; + + c_adv(msg->chn, offset); + *strm_off += offset; + +end: + DBG_TRACE_LEAVE(STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA|STRM_EV_FLT_ANA, s); + 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; + + DBG_TRACE_ENTER(STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA|STRM_EV_FLT_ANA, s, s->txn, msg); + list_for_each_entry(filter, &strm_flt(s)->filters, list) { + if (FLT_OPS(filter)->http_reset) { + DBG_TRACE_DEVEL(FLT_ID(filter), STRM_EV_HTTP_ANA|STRM_EV_FLT_ANA, s); + FLT_OPS(filter)->http_reset(s, filter, msg); + } + } + DBG_TRACE_LEAVE(STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA|STRM_EV_FLT_ANA, s); +} + +/* + * 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 buffer *msg) +{ + struct filter *filter; + + DBG_TRACE_ENTER(STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA|STRM_EV_FLT_ANA, s, s->txn, msg); + list_for_each_entry(filter, &strm_flt(s)->filters, list) { + if (FLT_OPS(filter)->http_reply) { + DBG_TRACE_DEVEL(FLT_ID(filter), STRM_EV_HTTP_ANA|STRM_EV_FLT_ANA, s); + FLT_OPS(filter)->http_reply(s, filter, status, msg); + } + } + DBG_TRACE_LEAVE(STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA|STRM_EV_FLT_ANA, s); +} + +/* + * Calls 'http_payload' 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 filters and the stream offset to be sure that a filter cannot + * forward more data than its predecessors. A filter can choose to not forward + * all data. Returns a negative value if an error occurs, else the number of + * forwarded bytes. + */ +int +flt_http_payload(struct stream *s, struct http_msg *msg, unsigned int len) +{ + struct filter *filter; + struct htx *htx; + unsigned long long *strm_off = &FLT_STRM_OFF(s, msg->chn); + unsigned int out = co_data(msg->chn); + int ret, data; + + strm_flt(s)->flags &= ~STRM_FLT_FL_HOLD_HTTP_HDRS; + + ret = data = len - out; + DBG_TRACE_ENTER(STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA|STRM_EV_FLT_ANA, s, s->txn, msg); + list_for_each_entry(filter, &strm_flt(s)->filters, list) { + unsigned long long *flt_off = &FLT_OFF(filter, msg->chn); + unsigned int offset = *flt_off - *strm_off; + + /* Call http_payload for filters only. Forward all data for + * others and update the filter offset + */ + if (!IS_DATA_FILTER(filter, msg->chn)) { + *flt_off += data - offset; + continue; + } + + if (FLT_OPS(filter)->http_payload) { + DBG_TRACE_DEVEL(FLT_ID(filter), STRM_EV_HTTP_ANA|STRM_EV_FLT_ANA, s); + ret = FLT_OPS(filter)->http_payload(s, filter, msg, out + offset, data - offset); + if (ret < 0) + goto end; + data = ret + *flt_off - *strm_off; + *flt_off += ret; + } + } + + /* If nothing was forwarded yet, we take care to hold the headers if + * following conditions are met : + * + * - *strm_off == 0 (nothing forwarded yet) + * - ret == 0 (no data forwarded at all on this turn) + * - STRM_FLT_FL_HOLD_HTTP_HDRS flag set (at least one filter want to hold the headers) + * + * Be careful, STRM_FLT_FL_HOLD_HTTP_HDRS is removed before each http_payload loop. + * Thus, it must explicitly be set when necessary. We must do that to hold the headers + * when there is no payload. + */ + if (!ret && !*strm_off && (strm_flt(s)->flags & STRM_FLT_FL_HOLD_HTTP_HDRS)) + goto end; + + ret = data; + *strm_off += ret; + end: + htx = htxbuf(&msg->chn->buf); + if (msg->flags & HTTP_MSGF_XFER_LEN) + htx->extra = 0; + DBG_TRACE_LEAVE(STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA|STRM_EV_FLT_ANA, s); + 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; + + DBG_TRACE_ENTER(STRM_EV_STRM_ANA|STRM_EV_FLT_ANA, s); + + /* 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. */ + + /* Set flag on channel to tell that the channel is filtered */ + chn->flags |= CF_FLT_ANALYZE; + chn->analysers |= ((chn->flags & CF_ISRESP) ? AN_RES_FLT_END : AN_REQ_FLT_END); + + RESUME_FILTER_LOOP(s, chn) { + if (!(chn->flags & CF_ISRESP)) { + if (an_bit == AN_REQ_FLT_START_BE && + !(filter->flags & FLT_FL_IS_BACKEND_FILTER)) + continue; + } + else { + if (an_bit == AN_RES_FLT_START_BE && + !(filter->flags & FLT_FL_IS_BACKEND_FILTER)) + continue; + } + + FLT_OFF(filter, chn) = 0; + if (FLT_OPS(filter)->channel_start_analyze) { + DBG_TRACE_DEVEL(FLT_ID(filter), STRM_EV_FLT_ANA, s); + ret = FLT_OPS(filter)->channel_start_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); + DBG_TRACE_LEAVE(STRM_EV_STRM_ANA|STRM_EV_FLT_ANA, s); + return ret; +} + +/* + * Calls 'channel_pre_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_pre_analyze' + * callback is resumable, so this function returns 0 if an error occurs or if it + * needs to wait, any other value otherwise. + * + * Note this function can be called many times for the same analyzer. In fact, + * it is called until the analyzer finishes its processing. + */ +int +flt_pre_analyze(struct stream *s, struct channel *chn, unsigned int an_bit) +{ + int ret = 1; + + DBG_TRACE_ENTER(STRM_EV_STRM_ANA|STRM_EV_FLT_ANA, s); + + RESUME_FILTER_LOOP(s, chn) { + if (FLT_OPS(filter)->channel_pre_analyze && (filter->pre_analyzers & an_bit)) { + DBG_TRACE_DEVEL(FLT_ID(filter), STRM_EV_FLT_ANA, s); + ret = FLT_OPS(filter)->channel_pre_analyze(s, filter, chn, an_bit); + if (ret <= 0) + BREAK_EXECUTION(s, chn, check_result); + filter->pre_analyzers &= ~an_bit; + } + } RESUME_FILTER_END; + + check_result: + ret = handle_analyzer_result(s, chn, 0, ret); + DBG_TRACE_LEAVE(STRM_EV_STRM_ANA|STRM_EV_FLT_ANA, s); + return ret; +} + +/* + * Calls 'channel_post_analyze' callback for all filters attached to a + * stream. This function is called AFTER each analyzer attached to a channel, + * expects analyzers responsible for data sending. 'channel_post_analyze' + * callback is NOT resumable, so this function returns a 0 if an error occurs, + * any other value otherwise. + * + * Here, AFTER means when the analyzer finishes its processing. + */ +int +flt_post_analyze(struct stream *s, struct channel *chn, unsigned int an_bit) +{ + struct filter *filter; + int ret = 1; + + DBG_TRACE_ENTER(STRM_EV_STRM_ANA|STRM_EV_FLT_ANA, s); + + list_for_each_entry(filter, &strm_flt(s)->filters, list) { + if (FLT_OPS(filter)->channel_post_analyze && (filter->post_analyzers & an_bit)) { + DBG_TRACE_DEVEL(FLT_ID(filter), STRM_EV_FLT_ANA, s); + ret = FLT_OPS(filter)->channel_post_analyze(s, filter, chn, an_bit); + if (ret < 0) + break; + filter->post_analyzers &= ~an_bit; + } + } + ret = handle_analyzer_result(s, chn, 0, ret); + DBG_TRACE_LEAVE(STRM_EV_STRM_ANA|STRM_EV_FLT_ANA, s); + return ret; +} + +/* + * This function is the AN_REQ/RES_FLT_HTTP_HDRS analyzer, used to filter HTTP + * headers or a request or a response. Returns 0 if an error occurs or if it + * needs to wait, any other value otherwise. + */ +int +flt_analyze_http_headers(struct stream *s, struct channel *chn, unsigned int an_bit) +{ + struct http_msg *msg; + int ret = 1; + + msg = ((chn->flags & CF_ISRESP) ? &s->txn->rsp : &s->txn->req); + DBG_TRACE_ENTER(STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA|STRM_EV_FLT_ANA, s, s->txn, msg); + + RESUME_FILTER_LOOP(s, chn) { + if (FLT_OPS(filter)->http_headers) { + DBG_TRACE_DEVEL(FLT_ID(filter), STRM_EV_HTTP_ANA|STRM_EV_FLT_ANA, s); + ret = FLT_OPS(filter)->http_headers(s, filter, msg); + if (ret <= 0) + BREAK_EXECUTION(s, chn, check_result); + } + } RESUME_FILTER_END; + + if (HAS_DATA_FILTERS(s, chn)) { + size_t data = http_get_hdrs_size(htxbuf(&chn->buf)); + struct filter *f; + + list_for_each_entry(f, &strm_flt(s)->filters, list) + FLT_OFF(f, chn) = data; + } + + check_result: + ret = handle_analyzer_result(s, chn, an_bit, ret); + DBG_TRACE_LEAVE(STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA|STRM_EV_FLT_ANA, s); + 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; + + DBG_TRACE_ENTER(STRM_EV_STRM_ANA|STRM_EV_FLT_ANA, s); + + /* Check if all filters attached on the stream have finished their + * processing on this channel. */ + if (!(chn->flags & CF_FLT_ANALYZE)) + goto sync; + + RESUME_FILTER_LOOP(s, chn) { + FLT_OFF(filter, chn) = 0; + unregister_data_filter(s, chn, filter); + + if (FLT_OPS(filter)->channel_end_analyze) { + DBG_TRACE_DEVEL(FLT_ID(filter), STRM_EV_FLT_ANA, s); + ret = FLT_OPS(filter)->channel_end_analyze(s, filter, chn); + if (ret <= 0) + BREAK_EXECUTION(s, chn, end); + } + } RESUME_FILTER_END; + + end: + /* We don't remove yet this analyzer because we need to synchronize the + * both channels. So here, we just remove the flag CF_FLT_ANALYZE. */ + ret = handle_analyzer_result(s, chn, 0, ret); + if (ret) { + chn->flags &= ~CF_FLT_ANALYZE; + + /* Pretend there is an activity on both channels. Flag on the + * current one will be automatically removed, so only the other + * one will remain. This is a way to be sure that + * 'channel_end_analyze' callback will have a chance to be + * called at least once for the other side to finish the current + * processing. Of course, this is the filter responsibility to + * wakeup the stream if it choose to loop on this callback. */ + s->req.flags |= CF_WAKE_ONCE; + s->res.flags |= CF_WAKE_ONCE; + } + + + sync: + /* Now we can check if filters have finished their work on the both + * channels */ + if (!(s->req.flags & CF_FLT_ANALYZE) && !(s->res.flags & CF_FLT_ANALYZE)) { + /* Sync channels by removing this analyzer for the both channels */ + s->req.analysers &= ~AN_REQ_FLT_END; + s->res.analysers &= ~AN_RES_FLT_END; + + /* Remove backend filters from the list */ + flt_stream_release(s, 1); + DBG_TRACE_LEAVE(STRM_EV_STRM_ANA|STRM_EV_FLT_ANA, s); + } + else { + DBG_TRACE_DEVEL("waiting for sync", STRM_EV_STRM_ANA|STRM_EV_FLT_ANA, s); + } + return ret; +} + + +/* + * Calls 'tcp_payload' callback for all "data" filters attached to a + * stream. This function is called when some data can be forwarded in the + * AN_REQ_FLT_XFER_BODY and AN_RES_FLT_XFER_BODY analyzers. It takes care to + * update the filters and the stream offset to be sure that a filter cannot + * forward more data than its predecessors. A filter can choose to not forward + * all data. Returns a negative value if an error occurs, else the number of + * forwarded bytes. + */ +int +flt_tcp_payload(struct stream *s, struct channel *chn, unsigned int len) +{ + struct filter *filter; + unsigned long long *strm_off = &FLT_STRM_OFF(s, chn); + unsigned int out = co_data(chn); + int ret, data; + + ret = data = len - out; + DBG_TRACE_ENTER(STRM_EV_TCP_ANA|STRM_EV_FLT_ANA, s); + list_for_each_entry(filter, &strm_flt(s)->filters, list) { + unsigned long long *flt_off = &FLT_OFF(filter, chn); + unsigned int offset = *flt_off - *strm_off; + + /* Call tcp_payload for filters only. Forward all data for + * others and update the filter offset + */ + if (!IS_DATA_FILTER(filter, chn)) { + *flt_off += data - offset; + continue; + } + + if (FLT_OPS(filter)->tcp_payload) { + + DBG_TRACE_DEVEL(FLT_ID(filter), STRM_EV_TCP_ANA|STRM_EV_FLT_ANA, s); + ret = FLT_OPS(filter)->tcp_payload(s, filter, chn, out + offset, data - offset); + if (ret < 0) + goto end; + data = ret + *flt_off - *strm_off; + *flt_off += ret; + } + } + + /* Only forward data if the last filter decides to forward something */ + if (ret > 0) { + ret = data; + *strm_off += ret; + } + end: + DBG_TRACE_LEAVE(STRM_EV_TCP_ANA|STRM_EV_FLT_ANA, s); + return ret; +} + +/* + * Called when TCP data must be filtered on a channel. This function is the + * AN_REQ/RES_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) +{ + unsigned int len; + int ret = 1; + + DBG_TRACE_ENTER(STRM_EV_STRM_ANA|STRM_EV_TCP_ANA|STRM_EV_FLT_ANA, s); + + /* If there is no "data" filters, we do nothing */ + if (!HAS_DATA_FILTERS(s, chn)) + goto end; + + if (s->flags & SF_HTX) { + struct htx *htx = htxbuf(&chn->buf); + len = htx->data; + } + else + len = c_data(chn); + + ret = flt_tcp_payload(s, chn, len); + if (ret < 0) + goto end; + c_adv(chn, ret); + + /* Stop waiting data if: + * - it the output is closed + * - the input in closed and no data is pending + * - There is a READ/WRITE timeout + */ + if (chn_cons(chn)->flags & SC_FL_SHUT_DONE) { + ret = 1; + goto end; + } + if (chn_prod(chn)->flags & (SC_FL_ABRT_DONE|SC_FL_EOS)) { + if (((s->flags & SF_HTX) && htx_is_empty(htxbuf(&chn->buf))) || c_empty(chn)) { + ret = 1; + goto end; + } + } + if (chn->flags & (CF_READ_TIMEOUT|CF_WRITE_TIMEOUT)) { + ret = 1; + goto end; + } + + /* Wait for data */ + DBG_TRACE_DEVEL("waiting for more data", STRM_EV_STRM_ANA|STRM_EV_TCP_ANA|STRM_EV_FLT_ANA, s); + return 0; + end: + /* Terminate the data filtering. If <ret> is negative, an error was + * encountered during the filtering. */ + ret = handle_analyzer_result(s, chn, an_bit, ret); + DBG_TRACE_LEAVE(STRM_EV_STRM_ANA|STRM_EV_TCP_ANA|STRM_EV_FLT_ANA, s); + return 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) +{ + 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 */ + if (IS_HTX_STRM(s)) { + http_set_term_flags(s); + + if (s->txn->status > 0) + http_reply_and_close(s, s->txn->status, NULL); + else { + s->txn->status = (!(chn->flags & CF_ISRESP)) ? 400 : 502; + http_reply_and_close(s, s->txn->status, http_error_message(s)); + } + } + else { + sess_set_term_flags(s); + stream_retnclose(s, NULL); + } + + if (!(chn->flags & CF_ISRESP)) + s->req.analysers &= AN_REQ_FLT_END; + else + s->res.analysers &= AN_RES_FLT_END; + + + DBG_TRACE_DEVEL("leaving on error", STRM_EV_FLT_ANA|STRM_EV_FLT_ERR, s); + return 0; + + wait: + if (!(chn->flags & CF_ISRESP)) + channel_dont_connect(chn); + DBG_TRACE_DEVEL("wairing for more data", STRM_EV_FLT_ANA, s); + 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 }, + } +}; + +INITCALL1(STG_REGISTER, cfg_register_keywords, &cfg_kws); + +REGISTER_POST_CHECK(flt_init_all); +REGISTER_PER_THREAD_INIT(flt_init_all_per_thread); +REGISTER_PER_THREAD_DEINIT(flt_deinit_all_per_thread); + +/* + * Local variables: + * c-indent-level: 8 + * c-basic-offset: 8 + * End: + */ |