lighttpd1.4/src/mod_proxy.c

#include "first.h"

#include "buffer.h"
#include "server.h"
#include "keyvalue.h"
#include "log.h"

#include "http_chunk.h"
#include "fdevent.h"
#include "inet_ntop_cache.h"
#include "connections.h"
#include "response.h"
#include "joblist.h"

#include "plugin.h"

#include "crc32.h"

#include <sys/types.h>

#include <unistd.h>
#include <errno.h>
#include <limits.h>
#include <string.h>
#include <stdlib.h>

#include "sys-socket.h"

#define data_proxy data_fastcgi
#define data_proxy_init data_fastcgi_init

#define PROXY_RETRY_TIMEOUT 60

/**
 *
 * HTTP reverse proxy
 *
 * TODO:      - HTTP/1.1
 *            - HTTP/1.1 persistent connection with upstream servers
 */

/* (future: might split struct and move part to http-header-glue.c) */
typedef struct http_header_remap_opts {
    const array *urlpaths;
    const array *hosts_request;
    const array *hosts_response;
    int https_remap;
    /*(not used in plugin_config, but used in handler_ctx)*/
    const buffer *http_host;
    const buffer *forwarded_host;
    const data_string *forwarded_urlpath;
} http_header_remap_opts;

typedef enum {
	PROXY_BALANCE_UNSET,
	PROXY_BALANCE_FAIR,
	PROXY_BALANCE_HASH,
	PROXY_BALANCE_RR,
	PROXY_BALANCE_STICKY
} proxy_balance_t;

typedef enum {
	PROXY_FORWARDED_NONE         = 0x00,
	PROXY_FORWARDED_FOR          = 0x01,
	PROXY_FORWARDED_PROTO        = 0x02,
	PROXY_FORWARDED_HOST         = 0x04,
	PROXY_FORWARDED_BY           = 0x08,
	PROXY_FORWARDED_REMOTE_USER  = 0x10
} proxy_forwarded_t;

typedef struct {
	array *extensions;
	array *forwarded_params;
	array *header_params;
	unsigned short debug;
	unsigned short replace_http_host;
	unsigned int forwarded;

	proxy_balance_t balance;
	http_header_remap_opts header;
} plugin_config;

typedef struct {
	PLUGIN_DATA;

	buffer *balance_buf;

	plugin_config **config_storage;

	plugin_config conf;
} plugin_data;

static int proxy_check_extforward;

typedef enum {
	PROXY_STATE_INIT,
	PROXY_STATE_CONNECT,
	PROXY_STATE_PREPARE_WRITE,
	PROXY_STATE_WRITE,
	PROXY_STATE_READ
} proxy_connection_state_t;

enum { PROXY_STDOUT, PROXY_END_REQUEST };

typedef struct {
	proxy_connection_state_t state;
	time_t state_timestamp;

	data_proxy *host;

	buffer *response;

	chunkqueue *wb;
	off_t wb_reqlen;

	int fd; /* fd to the proxy process */
	int fde_ndx; /* index into the fd-event buffer */

	http_response_opts opts;
	http_header_remap_opts remap_hdrs;
	plugin_config conf;

	connection *remote_conn;  /* dumb pointer */
	plugin_data *plugin_data; /* dumb pointer */
	data_array *ext;
} handler_ctx;

/* ok, we need a prototype */
static handler_t proxy_handle_fdevent(server *srv, void *ctx, int revents);

static handler_ctx * handler_ctx_init(void) {
	handler_ctx * hctx;


	hctx = calloc(1, sizeof(*hctx));

	hctx->state = PROXY_STATE_INIT;
	hctx->host = NULL;

	hctx->response = buffer_init();

	hctx->wb = chunkqueue_init();
	hctx->wb_reqlen = 0;

	hctx->fd = -1;
	hctx->fde_ndx = -1;

	return hctx;
}

static void handler_ctx_free(handler_ctx *hctx) {
	buffer_free(hctx->response);
	chunkqueue_free(hctx->wb);

	free(hctx);
}

INIT_FUNC(mod_proxy_init) {
	plugin_data *p;

	p = calloc(1, sizeof(*p));

	p->balance_buf = buffer_init();

	return p;
}


FREE_FUNC(mod_proxy_free) {
	plugin_data *p = p_d;

	UNUSED(srv);

	buffer_free(p->balance_buf);

	if (p->config_storage) {
		size_t i;
		for (i = 0; i < srv->config_context->used; i++) {
			plugin_config *s = p->config_storage[i];

			if (NULL == s) continue;

			array_free(s->extensions);
			array_free(s->forwarded_params);
			array_free(s->header_params);

			free(s);
		}
		free(p->config_storage);
	}

	free(p);

	return HANDLER_GO_ON;
}

SETDEFAULTS_FUNC(mod_proxy_set_defaults) {
	plugin_data *p = p_d;
	data_unset *du;
	size_t i = 0;

	config_values_t cv[] = {
		{ "proxy.server",              NULL, T_CONFIG_LOCAL, T_CONFIG_SCOPE_CONNECTION },       /* 0 */
		{ "proxy.debug",               NULL, T_CONFIG_SHORT, T_CONFIG_SCOPE_CONNECTION },       /* 1 */
		{ "proxy.balance",             NULL, T_CONFIG_STRING, T_CONFIG_SCOPE_CONNECTION },      /* 2 */
		{ "proxy.replace-http-host",   NULL, T_CONFIG_BOOLEAN, T_CONFIG_SCOPE_CONNECTION },     /* 3 */
		{ "proxy.forwarded",           NULL, T_CONFIG_ARRAY, T_CONFIG_SCOPE_CONNECTION },       /* 4 */
		{ "proxy.header",              NULL, T_CONFIG_ARRAY, T_CONFIG_SCOPE_CONNECTION },       /* 5 */
		{ NULL,                        NULL, T_CONFIG_UNSET, T_CONFIG_SCOPE_UNSET }
	};

	p->config_storage = calloc(1, srv->config_context->used * sizeof(plugin_config *));

	for (i = 0; i < srv->config_context->used; i++) {
		data_config const* config = (data_config const*)srv->config_context->data[i];
		plugin_config *s;

		s = calloc(1, sizeof(plugin_config));
		s->extensions    = array_init();
		s->debug         = 0;
		s->replace_http_host = 0;
		s->forwarded_params  = array_init();
		s->forwarded         = PROXY_FORWARDED_NONE;
		s->header_params     = array_init();

		cv[0].destination = s->extensions;
		cv[1].destination = &(s->debug);
		cv[2].destination = p->balance_buf;
		cv[3].destination = &(s->replace_http_host);
		cv[4].destination = s->forwarded_params;
		cv[5].destination = s->header_params;

		buffer_reset(p->balance_buf);

		p->config_storage[i] = s;

		if (0 != config_insert_values_global(srv, config->value, cv, i == 0 ? T_CONFIG_SCOPE_SERVER : T_CONFIG_SCOPE_CONNECTION)) {
			return HANDLER_ERROR;
		}

		if (buffer_string_is_empty(p->balance_buf)) {
			s->balance = PROXY_BALANCE_FAIR;
		} else if (buffer_is_equal_string(p->balance_buf, CONST_STR_LEN("fair"))) {
			s->balance = PROXY_BALANCE_FAIR;
		} else if (buffer_is_equal_string(p->balance_buf, CONST_STR_LEN("round-robin"))) {
			s->balance = PROXY_BALANCE_RR;
		} else if (buffer_is_equal_string(p->balance_buf, CONST_STR_LEN("hash"))) {
			s->balance = PROXY_BALANCE_HASH;
		} else if (buffer_is_equal_string(p->balance_buf, CONST_STR_LEN("sticky"))) {
					s->balance = PROXY_BALANCE_STICKY;
		} else {
			log_error_write(srv, __FILE__, __LINE__, "sb",
				        "proxy.balance has to be one of: fair, round-robin, hash, sticky, but not:", p->balance_buf);
			return HANDLER_ERROR;
		}

		if (!array_is_kvany(s->forwarded_params)) {
			log_error_write(srv, __FILE__, __LINE__, "s",
					"unexpected value for proxy.forwarded; expected ( \"param\" => \"value\" )");
			return HANDLER_ERROR;
		}
		for (size_t j = 0, used = s->forwarded_params->used; j < used; ++j) {
			proxy_forwarded_t param;
			du = s->forwarded_params->data[j];
			if (buffer_is_equal_string(du->key, CONST_STR_LEN("by"))) {
				param = PROXY_FORWARDED_BY;
			} else if (buffer_is_equal_string(du->key, CONST_STR_LEN("for"))) {
				param = PROXY_FORWARDED_FOR;
			} else if (buffer_is_equal_string(du->key, CONST_STR_LEN("host"))) {
				param = PROXY_FORWARDED_HOST;
			} else if (buffer_is_equal_string(du->key, CONST_STR_LEN("proto"))) {
				param = PROXY_FORWARDED_PROTO;
			} else if (buffer_is_equal_string(du->key, CONST_STR_LEN("remote_user"))) {
				param = PROXY_FORWARDED_REMOTE_USER;
			} else {
				log_error_write(srv, __FILE__, __LINE__, "sb",
					        "proxy.forwarded keys must be one of: by, for, host, proto, remote_user, but not:", du->key);
				return HANDLER_ERROR;
			}
			if (du->type == TYPE_STRING) {
				data_string *ds = (data_string *)du;
				if (buffer_is_equal_string(ds->value, CONST_STR_LEN("enable"))) {
					s->forwarded |= param;
				} else if (!buffer_is_equal_string(ds->value, CONST_STR_LEN("disable"))) {
					log_error_write(srv, __FILE__, __LINE__, "sb",
						        "proxy.forwarded values must be one of: 0, 1, enable, disable; error for key:", du->key);
					return HANDLER_ERROR;
				}
			} else if (du->type == TYPE_INTEGER) {
				data_integer *di = (data_integer *)du;
				if (di->value) s->forwarded |= param;
			} else {
				log_error_write(srv, __FILE__, __LINE__, "sb",
					        "proxy.forwarded values must be one of: 0, 1, enable, disable; error for key:", du->key);
				return HANDLER_ERROR;
			}
		}

		if (!array_is_kvany(s->header_params)) {
			log_error_write(srv, __FILE__, __LINE__, "s",
					"unexpected value for proxy.header; expected ( \"param\" => ( \"key\" => \"value\" ) )");
			return HANDLER_ERROR;
		}
		for (size_t j = 0, used = s->header_params->used; j < used; ++j) {
			data_array *da = (data_array *)s->header_params->data[j];
			if (buffer_is_equal_string(da->key, CONST_STR_LEN("https-remap"))) {
				data_string *ds = (data_string *)da;
				if (ds->type != TYPE_STRING) {
					log_error_write(srv, __FILE__, __LINE__, "s",
							"unexpected value for proxy.header; expected \"enable\" or \"disable\" for https-remap");
					return HANDLER_ERROR;
				}
				s->header.https_remap = !buffer_is_equal_string(ds->value, CONST_STR_LEN("disable"))
						     && !buffer_is_equal_string(ds->value, CONST_STR_LEN("0"));
				continue;
			}
			if (da->type != TYPE_ARRAY || !array_is_kvstring(da->value)) {
				log_error_write(srv, __FILE__, __LINE__, "sb",
						"unexpected value for proxy.header; expected ( \"param\" => ( \"key\" => \"value\" ) ) near key", da->key);
				return HANDLER_ERROR;
			}
			if (buffer_is_equal_string(da->key, CONST_STR_LEN("map-urlpath"))) {
				s->header.urlpaths = da->value;
			}
			else if (buffer_is_equal_string(da->key, CONST_STR_LEN("map-host-request"))) {
				s->header.hosts_request = da->value;
			}
			else if (buffer_is_equal_string(da->key, CONST_STR_LEN("map-host-response"))) {
				s->header.hosts_response = da->value;
			}
			else {
				log_error_write(srv, __FILE__, __LINE__, "sb",
						"unexpected key for proxy.header; expected ( \"param\" => ( \"key\" => \"value\" ) ) near key", da->key);
				return HANDLER_ERROR;
			}
		}

		if (NULL != (du = array_get_element(config->value, "proxy.server"))) {
			size_t j;
			data_array *da = (data_array *)du;

			if (du->type != TYPE_ARRAY || !array_is_kvarray(da->value)) {
				log_error_write(srv, __FILE__, __LINE__, "s",
						"unexpected value for proxy.server; expected ( \"ext\" => ( \"backend-label\" => ( \"key\" => \"value\" )))");

				return HANDLER_ERROR;
			}

			/*
			 * proxy.server = ( "<ext>" => ...,
			 *                  "<ext>" => ... )
			 */

			for (j = 0; j < da->value->used; j++) {
				data_array *da_ext = (data_array *)da->value->data[j];
				size_t n;

				/*
				 * proxy.server = ( "<ext>" =>
				 *                     ( "<host>" => ( ... ),
				 *                       "<host>" => ( ... )
				 *                     ),
				 *                    "<ext>" => ... )
				 */

				for (n = 0; n < da_ext->value->used; n++) {
					data_array *da_host = (data_array *)da_ext->value->data[n];

					data_proxy *df;
					data_array *dfa;

					config_values_t pcv[] = {
						{ "host",              NULL, T_CONFIG_STRING, T_CONFIG_SCOPE_CONNECTION },      /* 0 */
						{ "port",              NULL, T_CONFIG_SHORT, T_CONFIG_SCOPE_CONNECTION },       /* 1 */
						{ NULL,                NULL, T_CONFIG_UNSET, T_CONFIG_SCOPE_UNSET }
					};

					if (da_host->type != TYPE_ARRAY || !array_is_kvany(da_host->value)) {
						log_error_write(srv, __FILE__, __LINE__, "SBS",
								"unexpected value for proxy.server near [",
								da_host->key, "](string); expected ( \"ext\" => ( \"backend-label\" => ( \"key\" => \"value\" )))");

						return HANDLER_ERROR;
					}

					df = data_proxy_init();

					df->port = 80;

					buffer_copy_buffer(df->key, da_host->key);

					pcv[0].destination = df->host;
					pcv[1].destination = &(df->port);

					if (0 != config_insert_values_internal(srv, da_host->value, pcv, T_CONFIG_SCOPE_CONNECTION)) {
						df->free((data_unset*) df);
						return HANDLER_ERROR;
					}

					if (buffer_string_is_empty(df->host)) {
						log_error_write(srv, __FILE__, __LINE__, "sbbbs",
								"missing key (string):",
								da->key,
								da_ext->key,
								da_host->key,
								"host");

						df->free((data_unset*) df);
						return HANDLER_ERROR;
					}

					/* if extension already exists, take it */

					if (NULL == (dfa = (data_array *)array_get_element(s->extensions, da_ext->key->ptr))) {
						dfa = data_array_init();

						buffer_copy_buffer(dfa->key, da_ext->key);

						array_insert_unique(dfa->value, (data_unset *)df);
						array_insert_unique(s->extensions, (data_unset *)dfa);
					} else {
						array_insert_unique(dfa->value, (data_unset *)df);
					}
				}
			}
		}
	}

	for (i = 0; i < srv->srvconf.modules->used; i++) {
		data_string *ds = (data_string *)srv->srvconf.modules->data[i];
		if (buffer_is_equal_string(ds->value, CONST_STR_LEN("mod_extforward"))) {
			proxy_check_extforward = 1;
			break;
		}
	}

	return HANDLER_GO_ON;
}


static void proxy_backend_close(server *srv, handler_ctx *hctx) {
	if (hctx->fd != -1) {
		fdevent_event_del(srv->ev, &(hctx->fde_ndx), hctx->fd);
		fdevent_unregister(srv->ev, hctx->fd);
		fdevent_sched_close(srv->ev, hctx->fd, 1);
		hctx->fd = -1;
		hctx->fde_ndx = -1;
	}

	if (hctx->host) {
		hctx->host->usage--;
		hctx->host = NULL;
	}
}

static data_proxy * mod_proxy_extension_host_get(server *srv, connection *con, data_array *extension, proxy_balance_t balance, int debug) {
	unsigned long last_max = ULONG_MAX;
	int max_usage = INT_MAX;
	int ndx = -1;
	size_t k;

	if (extension->value->used == 1) {
		if ( ((data_proxy *)extension->value->data[0])->is_disabled ) {
			ndx = -1;
		} else {
			ndx = 0;
		}
	} else if (extension->value->used != 0) switch(balance) {
	case PROXY_BALANCE_HASH:
		/* hash balancing */

		if (debug) {
			log_error_write(srv, __FILE__, __LINE__,  "sd",
					"proxy - used hash balancing, hosts:", extension->value->used);
		}

		for (k = 0, ndx = -1, last_max = ULONG_MAX; k < extension->value->used; k++) {
			data_proxy *host = (data_proxy *)extension->value->data[k];
			unsigned long cur_max;

			if (host->is_disabled) continue;

			cur_max = generate_crc32c(CONST_BUF_LEN(con->uri.path)) +
				generate_crc32c(CONST_BUF_LEN(host->host)) + /* we can cache this */
				generate_crc32c(CONST_BUF_LEN(con->uri.authority));

			if (debug) {
				log_error_write(srv, __FILE__, __LINE__,  "sbbbd",
						"proxy - election:",
						con->uri.path,
						host->host,
						con->uri.authority,
						cur_max);
			}

			if ((last_max == ULONG_MAX) || /* first round */
			    (cur_max > last_max)) {
				last_max = cur_max;

				ndx = k;
			}
		}

		break;
	case PROXY_BALANCE_FAIR:
		/* fair balancing */
		if (debug) {
			log_error_write(srv, __FILE__, __LINE__,  "s",
					"proxy - used fair balancing");
		}

		for (k = 0, ndx = -1, max_usage = INT_MAX; k < extension->value->used; k++) {
			data_proxy *host = (data_proxy *)extension->value->data[k];

			if (host->is_disabled) continue;

			if (host->usage < max_usage) {
				max_usage = host->usage;

				ndx = k;
			}
		}

		break;
	case PROXY_BALANCE_RR: {
		data_proxy *host;

		/* round robin */
		if (debug) {
			log_error_write(srv, __FILE__, __LINE__,  "s",
					"proxy - used round-robin balancing");
		}

		/* just to be sure */
		force_assert(extension->value->used < INT_MAX);

		host = (data_proxy *)extension->value->data[0];

		/* Use last_used_ndx from first host in list */
		k = host->last_used_ndx;
		ndx = k + 1; /* use next host after the last one */
		if (ndx < 0) ndx = 0;

		/* Search first active host after last_used_ndx */
		while ( ndx < (int) extension->value->used
				&& (host = (data_proxy *)extension->value->data[ndx])->is_disabled ) ndx++;

		if (ndx >= (int) extension->value->used) {
			/* didn't found a higher id, wrap to the start */
			for (ndx = 0; ndx <= (int) k; ndx++) {
				host = (data_proxy *)extension->value->data[ndx];
				if (!host->is_disabled) break;
			}

			/* No active host found */
			if (host->is_disabled) ndx = -1;
		}

		/* Save new index for next round */
		((data_proxy *)extension->value->data[0])->last_used_ndx = ndx;

		break;
	}
	case PROXY_BALANCE_STICKY:
		/* source sticky balancing */

		if (debug) {
			log_error_write(srv, __FILE__, __LINE__,  "sd",
					"proxy - used sticky balancing, hosts:", extension->value->used);
		}

		for (k = 0, ndx = -1, last_max = ULONG_MAX; k < extension->value->used; k++) {
			data_proxy *host = (data_proxy *)extension->value->data[k];
			unsigned long cur_max;

			if (host->is_disabled) continue;

			cur_max = generate_crc32c(CONST_BUF_LEN(con->dst_addr_buf)) +
				generate_crc32c(CONST_BUF_LEN(host->host)) +
				host->port;

			if (debug) {
				log_error_write(srv, __FILE__, __LINE__,  "sbbdd",
						"proxy - election:",
						con->dst_addr_buf,
						host->host,
						host->port,
						cur_max);
			}

			if ((last_max == ULONG_MAX) || /* first round */
				(cur_max > last_max)) {
				last_max = cur_max;

				ndx = k;
			}
		}

		break;
	default:
		break;
	}

	/* found a server */
	if (ndx != -1) {
		data_proxy *host = (data_proxy *)extension->value->data[ndx];

		if (debug) {
			log_error_write(srv, __FILE__, __LINE__,  "sbd",
					"proxy - found a host",
					host->host, host->port);
		}

		host->usage++;
		return host;
	} else {
		/* no handler found */
		con->http_status = 503; /* Service Unavailable */
		con->mode = DIRECT;

		log_error_write(srv, __FILE__, __LINE__,  "sb",
				"no proxy-handler found for:",
				con->uri.path);

		return NULL;
	}
}

static void proxy_connection_close(server *srv, handler_ctx *hctx) {
	plugin_data *p;
	connection *con;

	p    = hctx->plugin_data;
	con  = hctx->remote_conn;

	proxy_backend_close(srv, hctx);
	handler_ctx_free(hctx);
	con->plugin_ctx[p->id] = NULL;

	/* finish response (if not already con->file_started, con->file_finished) */
	if (con->mode == p->id) {
		http_response_backend_done(srv, con);
	}
}

static handler_t proxy_reconnect(server *srv, handler_ctx *hctx) {
	proxy_backend_close(srv, hctx);

	hctx->host = mod_proxy_extension_host_get(srv, hctx->remote_conn, hctx->ext, hctx->conf.balance, (int)hctx->conf.debug);
	if (NULL == hctx->host) return HANDLER_FINISHED;

	hctx->state = PROXY_STATE_INIT;
	return HANDLER_COMEBACK;
}

static int proxy_establish_connection(server *srv, handler_ctx *hctx) {
	struct sockaddr *proxy_addr;
	struct sockaddr_in proxy_addr_in;
#if defined(HAVE_SYS_UN_H)
	struct sockaddr_un proxy_addr_un;
#endif
#if defined(HAVE_IPV6) && defined(HAVE_INET_PTON)
	struct sockaddr_in6 proxy_addr_in6;
#endif
	socklen_t servlen;

	data_proxy *host= hctx->host;
	int proxy_fd       = hctx->fd;


#if defined(HAVE_SYS_UN_H)
	if (strstr(host->host->ptr, "/")) {
		if (buffer_string_length(host->host) + 1 > sizeof(proxy_addr_un.sun_path)) {
			log_error_write(srv, __FILE__, __LINE__, "sB",
				"ERROR: Unix Domain socket filename too long:",
				host->host);