lighttpd1.4/src/mod_proxy.c

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

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

#include "plugin.h"

#include "inet_ntop_cache.h"
#include "crc32.h"

#include <sys/types.h>

#include <unistd.h>
#include <errno.h>
#include <fcntl.h>
#include <string.h>
#include <stdlib.h>
#include <ctype.h>
#include <assert.h>

#include <stdio.h>

#ifdef HAVE_SYS_FILIO_H
# include <sys/filio.h>
#endif

#include "sys-socket.h"

#define data_proxy data_fastcgi
#define data_proxy_init data_fastcgi_init

#define PROXY_RETRY_TIMEOUT 60

/**
 *
 * the proxy module is based on the fastcgi module
 *
 * 28.06.2004 Jan Kneschke     The first release
 * 01.07.2004 Evgeny Rodichev  Several bugfixes and cleanups
 *            - co-ordinate up- and downstream flows correctly (proxy_demux_response
 *              and proxy_handle_fdevent)
 *            - correctly transfer upstream http_response_status;
 *            - some unused structures removed.
 *
 * TODO:      - delay upstream read if write_queue is too large
 *              (to prevent memory eating, like in apache). Shoud be
 *              configurable).
 *            - persistent connection with upstream servers
 *            - HTTP/1.1
 */
typedef enum {
	PROXY_BALANCE_UNSET,
	PROXY_BALANCE_FAIR,
	PROXY_BALANCE_HASH,
	PROXY_BALANCE_RR
} proxy_balance_t;

typedef struct {
	array *extensions;
	unsigned short debug;

	proxy_balance_t balance;
} plugin_config;

typedef struct {
	PLUGIN_DATA;

	buffer *parse_response;
	buffer *balance_buf;

	plugin_config **config_storage;

	plugin_config conf;
} plugin_data;

typedef enum {
	PROXY_STATE_INIT,
	PROXY_STATE_CONNECT,
	PROXY_STATE_PREPARE_WRITE,
	PROXY_STATE_WRITE,
	PROXY_STATE_READ,
	PROXY_STATE_ERROR
} 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;
	buffer *response_header;

	chunkqueue *wb;

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

	size_t path_info_offset; /* start of path_info in uri.path */

	connection *remote_conn;  /* dump pointer */
	plugin_data *plugin_data; /* dump pointer */
} 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->response_header = buffer_init();

	hctx->wb = chunkqueue_init();

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

	return hctx;
}

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

	free(hctx);
}

INIT_FUNC(mod_proxy_init) {
	plugin_data *p;

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

	p->parse_response = buffer_init();
	p->balance_buf = buffer_init();

	return p;
}


FREE_FUNC(mod_proxy_free) {
	plugin_data *p = p_d;

	UNUSED(srv);

	buffer_free(p->parse_response);
	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 (s) {

				array_free(s->extensions);

				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 */
		{ 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++) {
		plugin_config *s;
		array *ca;

		s = malloc(sizeof(plugin_config));
		s->extensions    = array_init();
		s->debug         = 0;

		cv[0].destination = s->extensions;
		cv[1].destination = &(s->debug);
		cv[2].destination = p->balance_buf;

		buffer_reset(p->balance_buf);

		p->config_storage[i] = s;
		ca = ((data_config *)srv->config_context->data[i])->value;

		if (0 != config_insert_values_global(srv, ca, cv)) {
			return HANDLER_ERROR;
		}

		if (buffer_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 {
			log_error_write(srv, __FILE__, __LINE__, "sb",
				        "proxy.balance has to be one of: fair, round-robin, hash, but not:", p->balance_buf);
			return HANDLER_ERROR;
		}

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

			if (du->type != TYPE_ARRAY) {
				log_error_write(srv, __FILE__, __LINE__, "sss",
						"unexpected type for key: ", "proxy.server", "array of strings");

				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;

				if (da_ext->type != TYPE_ARRAY) {
					log_error_write(srv, __FILE__, __LINE__, "sssbs",
							"unexpected type for key: ", "proxy.server",
							"[", da->value->data[j]->key, "](string)");

					return HANDLER_ERROR;
				}

				/*
				 * 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) {
						log_error_write(srv, __FILE__, __LINE__, "ssSBS",
								"unexpected type for key:",
								"proxy.server",
								"[", da_ext->value->data[n]->key, "](string)");

						return HANDLER_ERROR;
					}

					df = data_proxy_init();

					df->port = 80;

					buffer_copy_string_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)) {
						df->free((data_unset*) df);
						return HANDLER_ERROR;
					}

					if (buffer_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_string_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);
					}
				}
			}
		}
	}

	return HANDLER_GO_ON;
}

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

	if (NULL == hctx) return;

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

	if (hctx->fd != -1) {
		fdevent_event_del(srv->ev, &(hctx->fde_ndx), hctx->fd);
		fdevent_unregister(srv->ev, hctx->fd);

		close(hctx->fd);
		srv->cur_fds--;
	}

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

	handler_ctx_free(hctx);
	con->plugin_ctx[p->id] = NULL;
}

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

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


#if defined(HAVE_IPV6) && defined(HAVE_INET_PTON)
	if (strstr(host->host->ptr, ":")) {
		memset(&proxy_addr_in6, 0, sizeof(proxy_addr_in6));
		proxy_addr_in6.sin6_family = AF_INET6;
		inet_pton(AF_INET6, host->host->ptr, (char *) &proxy_addr_in6.sin6_addr);
		proxy_addr_in6.sin6_port = htons(host->port);
		servlen = sizeof(proxy_addr_in6);
		proxy_addr = (struct sockaddr *) &proxy_addr_in6;
	} else
#endif
	{
		memset(&proxy_addr_in, 0, sizeof(proxy_addr_in));
		proxy_addr_in.sin_family = AF_INET;
		proxy_addr_in.sin_addr.s_addr = inet_addr(host->host->ptr);
		proxy_addr_in.sin_port = htons(host->port);
		servlen = sizeof(proxy_addr_in);
		proxy_addr = (struct sockaddr *) &proxy_addr_in;
	}


	if (-1 == connect(proxy_fd, proxy_addr, servlen)) {
		if (errno == EINPROGRESS || errno == EALREADY) {
			if (p->conf.debug) {
				log_error_write(srv, __FILE__, __LINE__, "sd",
						"connect delayed:", proxy_fd);
			}

			return 1;
		} else {

			log_error_write(srv, __FILE__, __LINE__, "sdsd",
					"connect failed:", proxy_fd, strerror(errno), errno);

			return -1;
		}
	}
	if (p->conf.debug) {
		log_error_write(srv, __FILE__, __LINE__, "sd",
				"connect succeeded: ", proxy_fd);
	}

	return 0;
}

static void proxy_set_header(connection *con, const char *key, const char *value) {
    data_string *ds_dst;

    if (NULL == (ds_dst = (data_string *)array_get_unused_element(con->request.headers, TYPE_STRING))) {
          ds_dst = data_string_init();
    }

    buffer_copy_string(ds_dst->key, key);
    buffer_copy_string(ds_dst->value, value);
    array_insert_unique(con->request.headers, (data_unset *)ds_dst);
}

static void proxy_append_header(connection *con, const char *key, const char *value) {
    data_string *ds_dst;

    if (NULL == (ds_dst = (data_string *)array_get_unused_element(con->request.headers, TYPE_STRING))) {
          ds_dst = data_string_init();
    }

    buffer_copy_string(ds_dst->key, key);
    buffer_append_string(ds_dst->value, value);
    array_insert_unique(con->request.headers, (data_unset *)ds_dst);
}


static int proxy_create_env(server *srv, handler_ctx *hctx) {
	size_t i;

	connection *con   = hctx->remote_conn;
	buffer *b;

	/* build header */

	b = chunkqueue_get_append_buffer(hctx->wb);

	/* request line */
	buffer_copy_string(b, get_http_method_name(con->request.http_method));
	buffer_append_string_len(b, CONST_STR_LEN(" "));

	buffer_append_string_buffer(b, con->request.uri);
	buffer_append_string_len(b, CONST_STR_LEN(" HTTP/1.0\r\n"));

	proxy_append_header(con, "X-Forwarded-For", (char *)inet_ntop_cache_get_ip(srv, &(con->dst_addr)));
	/* http_host is NOT is just a pointer to a buffer
	 * which is NULL if it is not set */
	if (con->request.http_host &&
	    !buffer_is_empty(con->request.http_host)) {
		proxy_set_header(con, "X-Host", con->request.http_host->ptr);
	}
	proxy_set_header(con, "X-Forwarded-Proto", con->uri.scheme->ptr);

	/* request header */
	for (i = 0; i < con->request.headers->used; i++) {
		data_string *ds;

		ds = (data_string *)con->request.headers->data[i];

		if (ds->value->used && ds->key->used) {
			if (buffer_is_equal_string(ds->key, CONST_STR_LEN("Connection"))) continue;
			if (buffer_is_equal_string(ds->key, CONST_STR_LEN("Proxy-Connection"))) continue;

			buffer_append_string_buffer(b, ds->key);
			buffer_append_string_len(b, CONST_STR_LEN(": "));
			buffer_append_string_buffer(b, ds->value);
			buffer_append_string_len(b, CONST_STR_LEN("\r\n"));
		}
	}

	buffer_append_string_len(b, CONST_STR_LEN("\r\n"));

	hctx->wb->bytes_in += b->used - 1;
	/* body */

	if (con->request.content_length) {
		chunkqueue *req_cq = con->request_content_queue;
		chunk *req_c;
		off_t offset;

		/* something to send ? */
		for (offset = 0, req_c = req_cq->first; offset != req_cq->bytes_in; req_c = req_c->next) {
			off_t weWant = req_cq->bytes_in - offset;
			off_t weHave = 0;

			/* we announce toWrite octects
			 * now take all the request_content chunk that we need to fill this request
			 * */

			switch (req_c->type) {
			case FILE_CHUNK:
				weHave = req_c->file.length - req_c->offset;

				if (weHave > weWant) weHave = weWant;

				chunkqueue_append_file(hctx->wb, req_c->file.name, req_c->offset, weHave);

				req_c->offset += weHave;
				req_cq->bytes_out += weHave;

				hctx->wb->bytes_in += weHave;

				break;
			case MEM_CHUNK:
				/* append to the buffer */
				weHave = req_c->mem->used - 1 - req_c->offset;

				if (weHave > weWant) weHave = weWant;

				b = chunkqueue_get_append_buffer(hctx->wb);
				buffer_append_memory(b, req_c->mem->ptr + req_c->offset, weHave);
				b->used++; /* add virtual \0 */

				req_c->offset += weHave;
				req_cq->bytes_out += weHave;

				hctx->wb->bytes_in += weHave;

				break;
			default:
				break;
			}

			offset += weHave;
		}

	}

	return 0;
}

static int proxy_set_state(server *srv, handler_ctx *hctx, proxy_connection_state_t state) {
	hctx->state = state;
	hctx->state_timestamp = srv->cur_ts;

	return 0;
}


static int proxy_response_parse(server *srv, connection *con, plugin_data *p, buffer *in) {
	char *s, *ns;
	int http_response_status = -1;

	UNUSED(srv);

	/* \r\n -> \0\0 */

	buffer_copy_string_buffer(p->parse_response, in);

	for (s = p->parse_response->ptr; NULL != (ns = strstr(s, "\r\n")); s = ns + 2) {
		char *key, *value;
		int key_len;
		data_string *ds;
		int copy_header;

		ns[0] = '\0';
		ns[1] = '\0';

		if (-1 == http_response_status) {
			/* The first line of a Response message is the Status-Line */

			for (key=s; *key && *key != ' '; key++);

			if (*key) {
				http_response_status = (int) strtol(key, NULL, 10);
				if (http_response_status <= 0) http_response_status = 502;
			} else {
				http_response_status = 502;
			}

			con->http_status = http_response_status;
			con->parsed_response |= HTTP_STATUS;
			continue;
		}

		if (NULL == (value = strchr(s, ':'))) {
			/* now we expect: "<key>: <value>\n" */

			continue;
		}

		key = s;
		key_len = value - key;

		value++;
		/* strip WS */
		while (*value == ' ' || *value == '\t') value++;

		copy_header = 1;

		switch(key_len) {
		case 4:
			if (0 == strncasecmp(key, "Date", key_len)) {
				con->parsed_response |= HTTP_DATE;
			}
			break;
		case 8:
			if (0 == strncasecmp(key, "Location", key_len)) {
				con->parsed_response |= HTTP_LOCATION;
			}
			break;
		case 10:
			if (0 == strncasecmp(key, "Connection", key_len)) {
				copy_header = 0;
			}
			break;
		case 14:
			if (0 == strncasecmp(key, "Content-Length", key_len)) {
				con->response.content_length = strtol(value, NULL, 10);
				con->parsed_response |= HTTP_CONTENT_LENGTH;
			}
			break;
		default:
			break;
		}

		if (copy_header) {
			if (NULL == (ds = (data_string *)array_get_unused_element(con->response.headers, TYPE_STRING))) {
				ds = data_response_init();
			}
			buffer_copy_string_len(ds->key, key, key_len);
			buffer_copy_string(ds->value, value);

			array_insert_unique(con->response.headers, (data_unset *)ds);
		}
	}

	return 0;
}


static int proxy_demux_response(server *srv, handler_ctx *hctx) {
	int fin = 0;
	int b;
	ssize_t r;

	plugin_data *p    = hctx->plugin_data;
	connection *con   = hctx->remote_conn;
	int proxy_fd       = hctx->fd;

	/* check how much we have to read */
	if (ioctl(hctx->fd, FIONREAD, &b)) {
		log_error_write(srv, __FILE__, __LINE__, "sd",
				"ioctl failed: ",
				proxy_fd);
		return -1;
	}


	if (p->conf.debug) {
		log_error_write(srv, __FILE__, __LINE__, "sd",
			       "proxy - have to read:", b);
	}

	if (b > 0) {
		if (hctx->response->used == 0) {
			/* avoid too small buffer */
			buffer_prepare_append(hctx->response, b + 1);
			hctx->response->used = 1;
		} else {
			buffer_prepare_append(hctx->response, b);
		}

		if (-1 == (r = read(hctx->fd, hctx->response->ptr + hctx->response->used - 1, b))) {
			if (errno == EAGAIN) return 0;
			log_error_write(srv, __FILE__, __LINE__, "sds",
					"unexpected end-of-file (perhaps the proxy process died):",
					proxy_fd, strerror(errno));
			return -1;
		}

		/* this should be catched by the b > 0 above */
		force_assert(r);

		hctx->response->used += r;
		hctx->response->ptr[hctx->response->used - 1] = '\0';

#if 0
		log_error_write(srv, __FILE__, __LINE__, "sdsbs",
				"demux: Response buffer len", hctx->response->used, ":", hctx->response, ":");
#endif

		if (0 == con->got_response) {
			con->got_response = 1;
			buffer_prepare_copy(hctx->response_header, 128);
		}