![[BACK]](/icons/cvsweb/back.gif){kind=icon}
Return to mod_proxy.c CVS log ![[TXT]](/icons/cvsweb/text.gif){kind=icon}
![[DIR]](/icons/cvsweb/dir.gif){kind=icon}
Up to [ELWIX - Embedded LightWeight unIX -] / embedaddon / lighttpd / src|
Return to mod_proxy.c CVS log | Up to [ELWIX - Embedded LightWeight unIX -] / embedaddon / lighttpd / src |
1.1 misho 1: #include "buffer.h"
2: #include "server.h"
3: #include "keyvalue.h"
4: #include "log.h"
5:
6: #include "http_chunk.h"
7: #include "fdevent.h"
8: #include "connections.h"
9: #include "response.h"
10: #include "joblist.h"
11:
12: #include "plugin.h"
13:
14: #include "inet_ntop_cache.h"
15: #include "crc32.h"
16:
17: #include <sys/types.h>
18:
19: #include <unistd.h>
20: #include <errno.h>
21: #include <fcntl.h>
22: #include <string.h>
23: #include <stdlib.h>
24: #include <ctype.h>
25: #include <assert.h>
26:
27: #include <stdio.h>
28:
29: #ifdef HAVE_SYS_FILIO_H
30: # include <sys/filio.h>
31: #endif
32:
33: #include "sys-socket.h"
34:
35: #define data_proxy data_fastcgi
36: #define data_proxy_init data_fastcgi_init
37:
38: #define PROXY_RETRY_TIMEOUT 60
39:
40: /**
41: *
42: * the proxy module is based on the fastcgi module
43: *
44: * 28.06.2004 Jan Kneschke The first release
45: * 01.07.2004 Evgeny Rodichev Several bugfixes and cleanups
46: * - co-ordinate up- and downstream flows correctly (proxy_demux_response
47: * and proxy_handle_fdevent)
48: * - correctly transfer upstream http_response_status;
49: * - some unused structures removed.
50: *
51: * TODO: - delay upstream read if write_queue is too large
52: * (to prevent memory eating, like in apache). Shoud be
53: * configurable).
54: * - persistent connection with upstream servers
55: * - HTTP/1.1
56: */
57: typedef enum {
58: PROXY_BALANCE_UNSET,
59: PROXY_BALANCE_FAIR,
60: PROXY_BALANCE_HASH,
61: PROXY_BALANCE_RR
62: } proxy_balance_t;
63:
64: typedef struct {
65: array *extensions;
66: unsigned short debug;
67:
68: proxy_balance_t balance;
69: } plugin_config;
70:
71: typedef struct {
72: PLUGIN_DATA;
73:
74: buffer *parse_response;
75: buffer *balance_buf;
76:
77: plugin_config **config_storage;
78:
79: plugin_config conf;
80: } plugin_data;
81:
82: typedef enum {
83: PROXY_STATE_INIT,
84: PROXY_STATE_CONNECT,
85: PROXY_STATE_PREPARE_WRITE,
86: PROXY_STATE_WRITE,
87: PROXY_STATE_READ,
88: PROXY_STATE_ERROR
89: } proxy_connection_state_t;
90:
91: enum { PROXY_STDOUT, PROXY_END_REQUEST };
92:
93: typedef struct {
94: proxy_connection_state_t state;
95: time_t state_timestamp;
96:
97: data_proxy *host;
98:
99: buffer *response;
100: buffer *response_header;
101:
102: chunkqueue *wb;
103:
104: int fd; /* fd to the proxy process */
105: int fde_ndx; /* index into the fd-event buffer */
106:
107: size_t path_info_offset; /* start of path_info in uri.path */
108:
109: connection *remote_conn; /* dump pointer */
110: plugin_data *plugin_data; /* dump pointer */
111: } handler_ctx;
112:
113:
114: /* ok, we need a prototype */
115: static handler_t proxy_handle_fdevent(server *srv, void *ctx, int revents);
116:
117: static handler_ctx * handler_ctx_init(void) {
118: handler_ctx * hctx;
119:
120:
121: hctx = calloc(1, sizeof(*hctx));
122:
123: hctx->state = PROXY_STATE_INIT;
124: hctx->host = NULL;
125:
126: hctx->response = buffer_init();
127: hctx->response_header = buffer_init();
128:
129: hctx->wb = chunkqueue_init();
130:
131: hctx->fd = -1;
132: hctx->fde_ndx = -1;
133:
134: return hctx;
135: }
136:
137: static void handler_ctx_free(handler_ctx *hctx) {
138: buffer_free(hctx->response);
139: buffer_free(hctx->response_header);
140: chunkqueue_free(hctx->wb);
141:
142: free(hctx);
143: }
144:
145: INIT_FUNC(mod_proxy_init) {
146: plugin_data *p;
147:
148: p = calloc(1, sizeof(*p));
149:
150: p->parse_response = buffer_init();
151: p->balance_buf = buffer_init();
152:
153: return p;
154: }
155:
156:
157: FREE_FUNC(mod_proxy_free) {
158: plugin_data *p = p_d;
159:
160: UNUSED(srv);
161:
162: buffer_free(p->parse_response);
163: buffer_free(p->balance_buf);
164:
165: if (p->config_storage) {
166: size_t i;
167: for (i = 0; i < srv->config_context->used; i++) {
168: plugin_config *s = p->config_storage[i];
169:
170: if (s) {
171:
172: array_free(s->extensions);
173:
174: free(s);
175: }
176: }
177: free(p->config_storage);
178: }
179:
180: free(p);
181:
182: return HANDLER_GO_ON;
183: }
184:
185: SETDEFAULTS_FUNC(mod_proxy_set_defaults) {
186: plugin_data *p = p_d;
187: data_unset *du;
188: size_t i = 0;
189:
190: config_values_t cv[] = {
191: { "proxy.server", NULL, T_CONFIG_LOCAL, T_CONFIG_SCOPE_CONNECTION }, /* 0 */
192: { "proxy.debug", NULL, T_CONFIG_SHORT, T_CONFIG_SCOPE_CONNECTION }, /* 1 */
193: { "proxy.balance", NULL, T_CONFIG_STRING, T_CONFIG_SCOPE_CONNECTION }, /* 2 */
194: { NULL, NULL, T_CONFIG_UNSET, T_CONFIG_SCOPE_UNSET }
195: };
196:
1.1.1.2 ! misho 197: p->config_storage = calloc(1, srv->config_context->used * sizeof(plugin_config *));
1.1 misho 198:
199: for (i = 0; i < srv->config_context->used; i++) {
200: plugin_config *s;
201: array *ca;
202:
203: s = malloc(sizeof(plugin_config));
204: s->extensions = array_init();
205: s->debug = 0;
206:
207: cv[0].destination = s->extensions;
208: cv[1].destination = &(s->debug);
209: cv[2].destination = p->balance_buf;
210:
211: buffer_reset(p->balance_buf);
212:
213: p->config_storage[i] = s;
214: ca = ((data_config *)srv->config_context->data[i])->value;
215:
216: if (0 != config_insert_values_global(srv, ca, cv)) {
217: return HANDLER_ERROR;
218: }
219:
220: if (buffer_is_empty(p->balance_buf)) {
221: s->balance = PROXY_BALANCE_FAIR;
222: } else if (buffer_is_equal_string(p->balance_buf, CONST_STR_LEN("fair"))) {
223: s->balance = PROXY_BALANCE_FAIR;
224: } else if (buffer_is_equal_string(p->balance_buf, CONST_STR_LEN("round-robin"))) {
225: s->balance = PROXY_BALANCE_RR;
226: } else if (buffer_is_equal_string(p->balance_buf, CONST_STR_LEN("hash"))) {
227: s->balance = PROXY_BALANCE_HASH;
228: } else {
229: log_error_write(srv, __FILE__, __LINE__, "sb",
230: "proxy.balance has to be one of: fair, round-robin, hash, but not:", p->balance_buf);
231: return HANDLER_ERROR;
232: }
233:
234: if (NULL != (du = array_get_element(ca, "proxy.server"))) {
235: size_t j;
236: data_array *da = (data_array *)du;
237:
238: if (du->type != TYPE_ARRAY) {
239: log_error_write(srv, __FILE__, __LINE__, "sss",
240: "unexpected type for key: ", "proxy.server", "array of strings");
241:
242: return HANDLER_ERROR;
243: }
244:
245: /*
246: * proxy.server = ( "<ext>" => ...,
247: * "<ext>" => ... )
248: */
249:
250: for (j = 0; j < da->value->used; j++) {
251: data_array *da_ext = (data_array *)da->value->data[j];
252: size_t n;
253:
254: if (da_ext->type != TYPE_ARRAY) {
255: log_error_write(srv, __FILE__, __LINE__, "sssbs",
256: "unexpected type for key: ", "proxy.server",
257: "[", da->value->data[j]->key, "](string)");
258:
259: return HANDLER_ERROR;
260: }
261:
262: /*
263: * proxy.server = ( "<ext>" =>
264: * ( "<host>" => ( ... ),
265: * "<host>" => ( ... )
266: * ),
267: * "<ext>" => ... )
268: */
269:
270: for (n = 0; n < da_ext->value->used; n++) {
271: data_array *da_host = (data_array *)da_ext->value->data[n];
272:
273: data_proxy *df;
274: data_array *dfa;
275:
276: config_values_t pcv[] = {
277: { "host", NULL, T_CONFIG_STRING, T_CONFIG_SCOPE_CONNECTION }, /* 0 */
278: { "port", NULL, T_CONFIG_SHORT, T_CONFIG_SCOPE_CONNECTION }, /* 1 */
279: { NULL, NULL, T_CONFIG_UNSET, T_CONFIG_SCOPE_UNSET }
280: };
281:
282: if (da_host->type != TYPE_ARRAY) {
283: log_error_write(srv, __FILE__, __LINE__, "ssSBS",
284: "unexpected type for key:",
285: "proxy.server",
286: "[", da_ext->value->data[n]->key, "](string)");
287:
288: return HANDLER_ERROR;
289: }
290:
291: df = data_proxy_init();
292:
293: df->port = 80;
294:
295: buffer_copy_string_buffer(df->key, da_host->key);
296:
297: pcv[0].destination = df->host;
298: pcv[1].destination = &(df->port);
299:
300: if (0 != config_insert_values_internal(srv, da_host->value, pcv)) {
1.1.1.2 ! misho 301: df->free((data_unset*) df);
1.1 misho 302: return HANDLER_ERROR;
303: }
304:
305: if (buffer_is_empty(df->host)) {
306: log_error_write(srv, __FILE__, __LINE__, "sbbbs",
307: "missing key (string):",
308: da->key,
309: da_ext->key,
310: da_host->key,
311: "host");
312:
1.1.1.2 ! misho 313: df->free((data_unset*) df);
1.1 misho 314: return HANDLER_ERROR;
315: }
316:
317: /* if extension already exists, take it */
318:
319: if (NULL == (dfa = (data_array *)array_get_element(s->extensions, da_ext->key->ptr))) {
320: dfa = data_array_init();
321:
322: buffer_copy_string_buffer(dfa->key, da_ext->key);
323:
324: array_insert_unique(dfa->value, (data_unset *)df);
325: array_insert_unique(s->extensions, (data_unset *)dfa);
326: } else {
327: array_insert_unique(dfa->value, (data_unset *)df);
328: }
329: }
330: }
331: }
332: }
333:
334: return HANDLER_GO_ON;
335: }
336:
337: static void proxy_connection_close(server *srv, handler_ctx *hctx) {
338: plugin_data *p;
339: connection *con;
340:
341: if (NULL == hctx) return;
342:
343: p = hctx->plugin_data;
344: con = hctx->remote_conn;
345:
346: if (hctx->fd != -1) {
347: fdevent_event_del(srv->ev, &(hctx->fde_ndx), hctx->fd);
348: fdevent_unregister(srv->ev, hctx->fd);
349:
350: close(hctx->fd);
351: srv->cur_fds--;
352: }
353:
354: if (hctx->host) {
355: hctx->host->usage--;
356: }
357:
358: handler_ctx_free(hctx);
359: con->plugin_ctx[p->id] = NULL;
360: }
361:
362: static int proxy_establish_connection(server *srv, handler_ctx *hctx) {
363: struct sockaddr *proxy_addr;
364: struct sockaddr_in proxy_addr_in;
365: #if defined(HAVE_IPV6) && defined(HAVE_INET_PTON)
366: struct sockaddr_in6 proxy_addr_in6;
367: #endif
368: socklen_t servlen;
369:
370: plugin_data *p = hctx->plugin_data;
371: data_proxy *host= hctx->host;
372: int proxy_fd = hctx->fd;
373:
374:
375: #if defined(HAVE_IPV6) && defined(HAVE_INET_PTON)
376: if (strstr(host->host->ptr, ":")) {
377: memset(&proxy_addr_in6, 0, sizeof(proxy_addr_in6));
378: proxy_addr_in6.sin6_family = AF_INET6;
379: inet_pton(AF_INET6, host->host->ptr, (char *) &proxy_addr_in6.sin6_addr);
380: proxy_addr_in6.sin6_port = htons(host->port);
381: servlen = sizeof(proxy_addr_in6);
382: proxy_addr = (struct sockaddr *) &proxy_addr_in6;
383: } else
384: #endif
385: {
386: memset(&proxy_addr_in, 0, sizeof(proxy_addr_in));
387: proxy_addr_in.sin_family = AF_INET;
388: proxy_addr_in.sin_addr.s_addr = inet_addr(host->host->ptr);
389: proxy_addr_in.sin_port = htons(host->port);
390: servlen = sizeof(proxy_addr_in);
391: proxy_addr = (struct sockaddr *) &proxy_addr_in;
392: }
393:
394:
395: if (-1 == connect(proxy_fd, proxy_addr, servlen)) {
396: if (errno == EINPROGRESS || errno == EALREADY) {
397: if (p->conf.debug) {
398: log_error_write(srv, __FILE__, __LINE__, "sd",
399: "connect delayed:", proxy_fd);
400: }
401:
402: return 1;
403: } else {
404:
405: log_error_write(srv, __FILE__, __LINE__, "sdsd",
406: "connect failed:", proxy_fd, strerror(errno), errno);
407:
408: return -1;
409: }
410: }
411: if (p->conf.debug) {
412: log_error_write(srv, __FILE__, __LINE__, "sd",
413: "connect succeeded: ", proxy_fd);
414: }
415:
416: return 0;
417: }
418:
419: static void proxy_set_header(connection *con, const char *key, const char *value) {
420: data_string *ds_dst;
421:
422: if (NULL == (ds_dst = (data_string *)array_get_unused_element(con->request.headers, TYPE_STRING))) {
423: ds_dst = data_string_init();
424: }
425:
426: buffer_copy_string(ds_dst->key, key);
427: buffer_copy_string(ds_dst->value, value);
428: array_insert_unique(con->request.headers, (data_unset *)ds_dst);
429: }
430:
431: static void proxy_append_header(connection *con, const char *key, const char *value) {
432: data_string *ds_dst;
433:
434: if (NULL == (ds_dst = (data_string *)array_get_unused_element(con->request.headers, TYPE_STRING))) {
435: ds_dst = data_string_init();
436: }
437:
438: buffer_copy_string(ds_dst->key, key);
439: buffer_append_string(ds_dst->value, value);
440: array_insert_unique(con->request.headers, (data_unset *)ds_dst);
441: }
442:
443:
444: static int proxy_create_env(server *srv, handler_ctx *hctx) {
445: size_t i;
446:
447: connection *con = hctx->remote_conn;
448: buffer *b;
449:
450: /* build header */
451:
452: b = chunkqueue_get_append_buffer(hctx->wb);
453:
454: /* request line */
455: buffer_copy_string(b, get_http_method_name(con->request.http_method));
456: buffer_append_string_len(b, CONST_STR_LEN(" "));
457:
458: buffer_append_string_buffer(b, con->request.uri);
459: buffer_append_string_len(b, CONST_STR_LEN(" HTTP/1.0\r\n"));
460:
461: proxy_append_header(con, "X-Forwarded-For", (char *)inet_ntop_cache_get_ip(srv, &(con->dst_addr)));
462: /* http_host is NOT is just a pointer to a buffer
463: * which is NULL if it is not set */
464: if (con->request.http_host &&
465: !buffer_is_empty(con->request.http_host)) {
466: proxy_set_header(con, "X-Host", con->request.http_host->ptr);
467: }
468: proxy_set_header(con, "X-Forwarded-Proto", con->uri.scheme->ptr);
469:
470: /* request header */
471: for (i = 0; i < con->request.headers->used; i++) {
472: data_string *ds;
473:
474: ds = (data_string *)con->request.headers->data[i];
475:
476: if (ds->value->used && ds->key->used) {
477: if (buffer_is_equal_string(ds->key, CONST_STR_LEN("Connection"))) continue;
478: if (buffer_is_equal_string(ds->key, CONST_STR_LEN("Proxy-Connection"))) continue;
479:
480: buffer_append_string_buffer(b, ds->key);
481: buffer_append_string_len(b, CONST_STR_LEN(": "));
482: buffer_append_string_buffer(b, ds->value);
483: buffer_append_string_len(b, CONST_STR_LEN("\r\n"));
484: }
485: }
486:
487: buffer_append_string_len(b, CONST_STR_LEN("\r\n"));
488:
489: hctx->wb->bytes_in += b->used - 1;
490: /* body */
491:
492: if (con->request.content_length) {
493: chunkqueue *req_cq = con->request_content_queue;
494: chunk *req_c;
495: off_t offset;
496:
497: /* something to send ? */
498: for (offset = 0, req_c = req_cq->first; offset != req_cq->bytes_in; req_c = req_c->next) {
499: off_t weWant = req_cq->bytes_in - offset;
500: off_t weHave = 0;
501:
502: /* we announce toWrite octects
503: * now take all the request_content chunk that we need to fill this request
504: * */
505:
506: switch (req_c->type) {
507: case FILE_CHUNK:
508: weHave = req_c->file.length - req_c->offset;
509:
510: if (weHave > weWant) weHave = weWant;
511:
512: chunkqueue_append_file(hctx->wb, req_c->file.name, req_c->offset, weHave);
513:
514: req_c->offset += weHave;
515: req_cq->bytes_out += weHave;
516:
517: hctx->wb->bytes_in += weHave;
518:
519: break;
520: case MEM_CHUNK:
521: /* append to the buffer */
522: weHave = req_c->mem->used - 1 - req_c->offset;
523:
524: if (weHave > weWant) weHave = weWant;
525:
526: b = chunkqueue_get_append_buffer(hctx->wb);
527: buffer_append_memory(b, req_c->mem->ptr + req_c->offset, weHave);
528: b->used++; /* add virtual \0 */
529:
530: req_c->offset += weHave;
531: req_cq->bytes_out += weHave;
532:
533: hctx->wb->bytes_in += weHave;
534:
535: break;
536: default:
537: break;
538: }
539:
540: offset += weHave;
541: }
542:
543: }
544:
545: return 0;
546: }
547:
548: static int proxy_set_state(server *srv, handler_ctx *hctx, proxy_connection_state_t state) {
549: hctx->state = state;
550: hctx->state_timestamp = srv->cur_ts;
551:
552: return 0;
553: }
554:
555:
556: static int proxy_response_parse(server *srv, connection *con, plugin_data *p, buffer *in) {
557: char *s, *ns;
558: int http_response_status = -1;
559:
560: UNUSED(srv);
561:
562: /* \r\n -> \0\0 */
563:
564: buffer_copy_string_buffer(p->parse_response, in);
565:
566: for (s = p->parse_response->ptr; NULL != (ns = strstr(s, "\r\n")); s = ns + 2) {
567: char *key, *value;
568: int key_len;
569: data_string *ds;
570: int copy_header;
571:
572: ns[0] = '\0';
573: ns[1] = '\0';
574:
575: if (-1 == http_response_status) {
576: /* The first line of a Response message is the Status-Line */
577:
578: for (key=s; *key && *key != ' '; key++);
579:
580: if (*key) {
581: http_response_status = (int) strtol(key, NULL, 10);
582: if (http_response_status <= 0) http_response_status = 502;
583: } else {
584: http_response_status = 502;
585: }
586:
587: con->http_status = http_response_status;
588: con->parsed_response |= HTTP_STATUS;
589: continue;
590: }
591:
592: if (NULL == (value = strchr(s, ':'))) {
593: /* now we expect: "<key>: <value>\n" */
594:
595: continue;
596: }
597:
598: key = s;
599: key_len = value - key;
600:
601: value++;
602: /* strip WS */
603: while (*value == ' ' || *value == '\t') value++;
604:
605: copy_header = 1;
606:
607: switch(key_len) {
608: case 4:
609: if (0 == strncasecmp(key, "Date", key_len)) {
610: con->parsed_response |= HTTP_DATE;
611: }
612: break;
613: case 8:
614: if (0 == strncasecmp(key, "Location", key_len)) {
615: con->parsed_response |= HTTP_LOCATION;
616: }
617: break;
618: case 10:
619: if (0 == strncasecmp(key, "Connection", key_len)) {
620: copy_header = 0;
621: }
622: break;
623: case 14:
624: if (0 == strncasecmp(key, "Content-Length", key_len)) {
625: con->response.content_length = strtol(value, NULL, 10);
626: con->parsed_response |= HTTP_CONTENT_LENGTH;
627: }
628: break;
629: default:
630: break;
631: }
632:
633: if (copy_header) {
634: if (NULL == (ds = (data_string *)array_get_unused_element(con->response.headers, TYPE_STRING))) {
635: ds = data_response_init();
636: }
637: buffer_copy_string_len(ds->key, key, key_len);
638: buffer_copy_string(ds->value, value);
639:
640: array_insert_unique(con->response.headers, (data_unset *)ds);
641: }
642: }
643:
644: return 0;
645: }
646:
647:
648: static int proxy_demux_response(server *srv, handler_ctx *hctx) {
649: int fin = 0;
650: int b;
651: ssize_t r;
652:
653: plugin_data *p = hctx->plugin_data;
654: connection *con = hctx->remote_conn;
655: int proxy_fd = hctx->fd;
656:
657: /* check how much we have to read */
658: if (ioctl(hctx->fd, FIONREAD, &b)) {
659: log_error_write(srv, __FILE__, __LINE__, "sd",
660: "ioctl failed: ",
661: proxy_fd);
662: return -1;
663: }
664:
665:
666: if (p->conf.debug) {
667: log_error_write(srv, __FILE__, __LINE__, "sd",
668: "proxy - have to read:", b);
669: }
670:
671: if (b > 0) {
672: if (hctx->response->used == 0) {
673: /* avoid too small buffer */
674: buffer_prepare_append(hctx->response, b + 1);
675: hctx->response->used = 1;
676: } else {
677: buffer_prepare_append(hctx->response, b);
678: }
679:
680: if (-1 == (r = read(hctx->fd, hctx->response->ptr + hctx->response->used - 1, b))) {
681: if (errno == EAGAIN) return 0;
682: log_error_write(srv, __FILE__, __LINE__, "sds",
683: "unexpected end-of-file (perhaps the proxy process died):",
684: proxy_fd, strerror(errno));
685: return -1;
686: }
687:
688: /* this should be catched by the b > 0 above */
1.1.1.2 ! misho 689: force_assert(r);
1.1 misho 690:
691: hctx->response->used += r;
692: hctx->response->ptr[hctx->response->used - 1] = '\0';
693:
694: #if 0
695: log_error_write(srv, __FILE__, __LINE__, "sdsbs",
696: "demux: Response buffer len", hctx->response->used, ":", hctx->response, ":");
697: #endif
698:
699: if (0 == con->got_response) {
700: con->got_response = 1;
701: buffer_prepare_copy(hctx->response_header, 128);
702: }
703:
704: if (0 == con->file_started) {
705: char *c;
706:
707: /* search for the \r\n\r\n in the string */
708: if (NULL != (c = buffer_search_string_len(hctx->response, "\r\n\r\n", 4))) {
709: size_t hlen = c - hctx->response->ptr + 4;
710: size_t blen = hctx->response->used - hlen - 1;
711: /* found */
712:
713: buffer_append_string_len(hctx->response_header, hctx->response->ptr, c - hctx->response->ptr + 4);
714: #if 0
715: log_error_write(srv, __FILE__, __LINE__, "sb", "Header:", hctx->response_header);
716: #endif
717: /* parse the response header */
718: proxy_response_parse(srv, con, p, hctx->response_header);
719:
720: /* enable chunked-transfer-encoding */
721: if (con->request.http_version == HTTP_VERSION_1_1 &&
722: !(con->parsed_response & HTTP_CONTENT_LENGTH)) {
723: con->response.transfer_encoding = HTTP_TRANSFER_ENCODING_CHUNKED;
724: }
725:
726: con->file_started = 1;
727: if (blen) {
728: http_chunk_append_mem(srv, con, c + 4, blen + 1);
729: }
730: hctx->response->used = 0;
731: joblist_append(srv, con);
732: }
733: } else {
734: http_chunk_append_mem(srv, con, hctx->response->ptr, hctx->response->used);
735: joblist_append(srv, con);
736: hctx->response->used = 0;
737: }
738:
739: } else {
740: /* reading from upstream done */
741: con->file_finished = 1;
742:
743: http_chunk_append_mem(srv, con, NULL, 0);
744: joblist_append(srv, con);
745:
746: fin = 1;
747: }
748:
749: return fin;
750: }
751:
752:
753: static handler_t proxy_write_request(server *srv, handler_ctx *hctx) {
754: data_proxy *host= hctx->host;
755: connection *con = hctx->remote_conn;
756:
757: int ret;
758:
759: if (!host ||
760: (!host->host->used || !host->port)) return -1;
761:
762: switch(hctx->state) {
763: case PROXY_STATE_CONNECT:
764: /* wait for the connect() to finish */
765:
766: /* connect failed ? */
767: if (-1 == hctx->fde_ndx) return HANDLER_ERROR;
768:
769: /* wait */
770: return HANDLER_WAIT_FOR_EVENT;
771:
772: break;
773:
774: case PROXY_STATE_INIT:
775: #if defined(HAVE_IPV6) && defined(HAVE_INET_PTON)
776: if (strstr(host->host->ptr,":")) {
777: if (-1 == (hctx->fd = socket(AF_INET6, SOCK_STREAM, 0))) {
778: log_error_write(srv, __FILE__, __LINE__, "ss", "socket failed: ", strerror(errno));
779: return HANDLER_ERROR;
780: }
781: } else
782: #endif
783: {
784: if (-1 == (hctx->fd = socket(AF_INET, SOCK_STREAM, 0))) {
785: log_error_write(srv, __FILE__, __LINE__, "ss", "socket failed: ", strerror(errno));
786: return HANDLER_ERROR;
787: }
788: }
789: hctx->fde_ndx = -1;
790:
791: srv->cur_fds++;
792:
793: fdevent_register(srv->ev, hctx->fd, proxy_handle_fdevent, hctx);
794:
795: if (-1 == fdevent_fcntl_set(srv->ev, hctx->fd)) {
796: log_error_write(srv, __FILE__, __LINE__, "ss", "fcntl failed: ", strerror(errno));
797:
798: return HANDLER_ERROR;
799: }
800:
801: switch (proxy_establish_connection(srv, hctx)) {
802: case 1:
803: proxy_set_state(srv, hctx, PROXY_STATE_CONNECT);
804:
805: /* connection is in progress, wait for an event and call getsockopt() below */
806:
807: fdevent_event_set(srv->ev, &(hctx->fde_ndx), hctx->fd, FDEVENT_OUT);
808:
809: return HANDLER_WAIT_FOR_EVENT;
810: case -1:
811: /* if ECONNREFUSED choose another connection -> FIXME */
812: hctx->fde_ndx = -1;
813:
814: return HANDLER_ERROR;
815: default:
816: /* everything is ok, go on */
817: proxy_set_state(srv, hctx, PROXY_STATE_PREPARE_WRITE);
818: break;
819: }
820:
821: /* fall through */
822:
823: case PROXY_STATE_PREPARE_WRITE:
824: proxy_create_env(srv, hctx);
825:
826: proxy_set_state(srv, hctx, PROXY_STATE_WRITE);
827:
828: /* fall through */
829: case PROXY_STATE_WRITE:;
830: ret = srv->network_backend_write(srv, con, hctx->fd, hctx->wb, MAX_WRITE_LIMIT);
831:
832: chunkqueue_remove_finished_chunks(hctx->wb);
833:
834: if (-1 == ret) { /* error on our side */
835: log_error_write(srv, __FILE__, __LINE__, "ssd", "write failed:", strerror(errno), errno);
836:
837: return HANDLER_ERROR;
838: } else if (-2 == ret) { /* remote close */
839: log_error_write(srv, __FILE__, __LINE__, "ssd", "write failed, remote connection close:", strerror(errno), errno);
840:
841: return HANDLER_ERROR;
842: }
843:
844: if (hctx->wb->bytes_out == hctx->wb->bytes_in) {
845: proxy_set_state(srv, hctx, PROXY_STATE_READ);
846:
847: fdevent_event_del(srv->ev, &(hctx->fde_ndx), hctx->fd);
848: fdevent_event_set(srv->ev, &(hctx->fde_ndx), hctx->fd, FDEVENT_IN);
849: } else {
850: fdevent_event_set(srv->ev, &(hctx->fde_ndx), hctx->fd, FDEVENT_OUT);
851:
852: return HANDLER_WAIT_FOR_EVENT;
853: }
854:
855: return HANDLER_WAIT_FOR_EVENT;
856: case PROXY_STATE_READ:
857: /* waiting for a response */
858: return HANDLER_WAIT_FOR_EVENT;
859: default:
860: log_error_write(srv, __FILE__, __LINE__, "s", "(debug) unknown state");
861: return HANDLER_ERROR;
862: }
863:
864: return HANDLER_GO_ON;
865: }
866:
867: #define PATCH(x) \
868: p->conf.x = s->x;
869: static int mod_proxy_patch_connection(server *srv, connection *con, plugin_data *p) {
870: size_t i, j;
871: plugin_config *s = p->config_storage[0];
872:
873: PATCH(extensions);
874: PATCH(debug);
875: PATCH(balance);
876:
877: /* skip the first, the global context */
878: for (i = 1; i < srv->config_context->used; i++) {
879: data_config *dc = (data_config *)srv->config_context->data[i];
880: s = p->config_storage[i];
881:
882: /* condition didn't match */
883: if (!config_check_cond(srv, con, dc)) continue;
884:
885: /* merge config */
886: for (j = 0; j < dc->value->used; j++) {
887: data_unset *du = dc->value->data[j];
888:
889: if (buffer_is_equal_string(du->key, CONST_STR_LEN("proxy.server"))) {
890: PATCH(extensions);
891: } else if (buffer_is_equal_string(du->key, CONST_STR_LEN("proxy.debug"))) {
892: PATCH(debug);
893: } else if (buffer_is_equal_string(du->key, CONST_STR_LEN("proxy.balance"))) {
894: PATCH(balance);
895: }
896: }
897: }
898:
899: return 0;
900: }
901: #undef PATCH
902:
903: SUBREQUEST_FUNC(mod_proxy_handle_subrequest) {
904: plugin_data *p = p_d;
905:
906: handler_ctx *hctx = con->plugin_ctx[p->id];
907: data_proxy *host;
908:
909: if (NULL == hctx) return HANDLER_GO_ON;
910:
911: mod_proxy_patch_connection(srv, con, p);
912:
913: host = hctx->host;
914:
915: /* not my job */
916: if (con->mode != p->id) return HANDLER_GO_ON;
917:
918: /* ok, create the request */
919: switch(proxy_write_request(srv, hctx)) {
920: case HANDLER_ERROR:
921: log_error_write(srv, __FILE__, __LINE__, "sbdd", "proxy-server disabled:",
922: host->host,
923: host->port,
924: hctx->fd);
925:
926: /* disable this server */
927: host->is_disabled = 1;
928: host->disable_ts = srv->cur_ts;
929:
930: proxy_connection_close(srv, hctx);
931:
932: /* reset the enviroment and restart the sub-request */
933: buffer_reset(con->physical.path);
934: con->mode = DIRECT;
935:
936: joblist_append(srv, con);
937:
938: /* mis-using HANDLER_WAIT_FOR_FD to break out of the loop
939: * and hope that the childs will be restarted
940: *
941: */
942:
943: return HANDLER_WAIT_FOR_FD;
944: case HANDLER_WAIT_FOR_EVENT:
945: break;
946: case HANDLER_WAIT_FOR_FD:
947: return HANDLER_WAIT_FOR_FD;
948: default:
949: break;
950: }
951:
952: if (con->file_started == 1) {
953: return HANDLER_FINISHED;
954: } else {
955: return HANDLER_WAIT_FOR_EVENT;
956: }
957: }
958:
959: static handler_t proxy_handle_fdevent(server *srv, void *ctx, int revents) {
960: handler_ctx *hctx = ctx;
961: connection *con = hctx->remote_conn;
962: plugin_data *p = hctx->plugin_data;
963:
964:
965: if ((revents & FDEVENT_IN) &&
966: hctx->state == PROXY_STATE_READ) {
967:
968: if (p->conf.debug) {
969: log_error_write(srv, __FILE__, __LINE__, "sd",
970: "proxy: fdevent-in", hctx->state);
971: }
972:
973: switch (proxy_demux_response(srv, hctx)) {
974: case 0:
975: break;
976: case 1:
977: /* we are done */
978: proxy_connection_close(srv, hctx);
979:
980: joblist_append(srv, con);
981: return HANDLER_FINISHED;
982: case -1:
983: if (con->file_started == 0) {
984: /* nothing has been send out yet, send a 500 */
985: connection_set_state(srv, con, CON_STATE_HANDLE_REQUEST);
986: con->http_status = 500;
987: con->mode = DIRECT;
988: } else {
989: /* response might have been already started, kill the connection */
990: connection_set_state(srv, con, CON_STATE_ERROR);
991: }
992:
993: joblist_append(srv, con);
994: return HANDLER_FINISHED;
995: }
996: }
997:
998: if (revents & FDEVENT_OUT) {
999: if (p->conf.debug) {
1000: log_error_write(srv, __FILE__, __LINE__, "sd",
1001: "proxy: fdevent-out", hctx->state);
1002: }
1003:
1004: if (hctx->state == PROXY_STATE_CONNECT) {
1005: int socket_error;
1006: socklen_t socket_error_len = sizeof(socket_error);
1007:
1008: /* we don't need it anymore */
1009: fdevent_event_del(srv->ev, &(hctx->fde_ndx), hctx->fd);
1010: hctx->fde_ndx = -1;
1011:
1012: /* try to finish the connect() */
1013: if (0 != getsockopt(hctx->fd, SOL_SOCKET, SO_ERROR, &socket_error, &socket_error_len)) {
1014: log_error_write(srv, __FILE__, __LINE__, "ss",
1015: "getsockopt failed:", strerror(errno));
1016:
1017: joblist_append(srv, con);
1018: return HANDLER_FINISHED;
1019: }
1020: if (socket_error != 0) {
1021: log_error_write(srv, __FILE__, __LINE__, "ss",
1022: "establishing connection failed:", strerror(socket_error),
1023: "port:", hctx->host->port);
1024:
1025: joblist_append(srv, con);
1026: return HANDLER_FINISHED;
1027: }
1028: if (p->conf.debug) {
1029: log_error_write(srv, __FILE__, __LINE__, "s", "proxy - connect - delayed success");
1030: }
1031:
1032: proxy_set_state(srv, hctx, PROXY_STATE_PREPARE_WRITE);
1033: }
1034:
1035: if (hctx->state == PROXY_STATE_PREPARE_WRITE ||
1036: hctx->state == PROXY_STATE_WRITE) {
1037: /* we are allowed to send something out
1038: *
1039: * 1. after a just finished connect() call
1040: * 2. in a unfinished write() call (long POST request)
1041: */
1042: return mod_proxy_handle_subrequest(srv, con, p);
1043: } else {
1044: log_error_write(srv, __FILE__, __LINE__, "sd",
1045: "proxy: out", hctx->state);
1046: }
1047: }
1048:
1049: /* perhaps this issue is already handled */
1050: if (revents & FDEVENT_HUP) {
1051: if (p->conf.debug) {
1052: log_error_write(srv, __FILE__, __LINE__, "sd",
1053: "proxy: fdevent-hup", hctx->state);
1054: }
1055:
1056: if (hctx->state == PROXY_STATE_CONNECT) {
1057: /* connect() -> EINPROGRESS -> HUP */
1058:
1059: /**
1060: * what is proxy is doing if it can't reach the next hop ?
1061: *
1062: */
1063:
1064: if (hctx->host) {
1065: hctx->host->is_disabled = 1;
1066: hctx->host->disable_ts = srv->cur_ts;
1067: log_error_write(srv, __FILE__, __LINE__, "sbdd", "proxy-server disabled:",
1068: hctx->host->host,
1069: hctx->host->port,
1070: hctx->fd);
1071:
1072: /* disable this server */
1073: hctx->host->is_disabled = 1;
1074: hctx->host->disable_ts = srv->cur_ts;
1075:
1076: proxy_connection_close(srv, hctx);
1077:
1078: /* reset the enviroment and restart the sub-request */
1079: buffer_reset(con->physical.path);
1080: con->mode = DIRECT;
1081:
1082: joblist_append(srv, con);
1083: } else {
1084: proxy_connection_close(srv, hctx);
1085: joblist_append(srv, con);
1086:
1087: con->mode = DIRECT;
1088: con->http_status = 503;
1089: }
1090:
1091: return HANDLER_FINISHED;
1092: }
1093:
1094: if (!con->file_finished) {
1095: http_chunk_append_mem(srv, con, NULL, 0);
1096: }
1097:
1098: con->file_finished = 1;
1099: proxy_connection_close(srv, hctx);
1100: joblist_append(srv, con);
1101: } else if (revents & FDEVENT_ERR) {
1102: /* kill all connections to the proxy process */
1103:
1104: log_error_write(srv, __FILE__, __LINE__, "sd", "proxy-FDEVENT_ERR, but no HUP", revents);
1105:
1106: con->file_finished = 1;
1107: joblist_append(srv, con);
1108: proxy_connection_close(srv, hctx);
1109: }
1110:
1111: return HANDLER_FINISHED;
1112: }
1113:
1114: static handler_t mod_proxy_check_extension(server *srv, connection *con, void *p_d) {
1115: plugin_data *p = p_d;
1116: size_t s_len;
1117: unsigned long last_max = ULONG_MAX;
1118: int max_usage = INT_MAX;
1119: int ndx = -1;
1120: size_t k;
1121: buffer *fn;
1122: data_array *extension = NULL;
1123: size_t path_info_offset;
1124:
1125: if (con->mode != DIRECT) return HANDLER_GO_ON;
1126:
1127: /* Possibly, we processed already this request */
1128: if (con->file_started == 1) return HANDLER_GO_ON;
1129:
1130: mod_proxy_patch_connection(srv, con, p);
1131:
1132: fn = con->uri.path;
1133:
1134: if (fn->used == 0) {
1135: return HANDLER_ERROR;
1136: }
1137:
1138: s_len = fn->used - 1;
1139:
1140:
1141: path_info_offset = 0;
1142:
1143: if (p->conf.debug) {
1144: log_error_write(srv, __FILE__, __LINE__, "s", "proxy - start");
1145: }
1146:
1147: /* check if extension matches */
1148: for (k = 0; k < p->conf.extensions->used; k++) {
1149: data_array *ext = NULL;
1150: size_t ct_len;
1151:
1152: ext = (data_array *)p->conf.extensions->data[k];
1153:
1154: if (ext->key->used == 0) continue;
1155:
1156: ct_len = ext->key->used - 1;
1157:
1158: if (s_len < ct_len) continue;
1159:
1160: /* check extension in the form "/proxy_pattern" */
1161: if (*(ext->key->ptr) == '/') {
1162: if (strncmp(fn->ptr, ext->key->ptr, ct_len) == 0) {
1163: if (s_len > ct_len + 1) {
1164: char *pi_offset;
1165:
1166: if (NULL != (pi_offset = strchr(fn->ptr + ct_len + 1, '/'))) {
1167: path_info_offset = pi_offset - fn->ptr;
1168: }
1169: }
1170: extension = ext;
1171: break;
1172: }
1173: } else if (0 == strncmp(fn->ptr + s_len - ct_len, ext->key->ptr, ct_len)) {
1174: /* check extension in the form ".fcg" */
1175: extension = ext;
1176: break;
1177: }
1178: }
1179:
1180: if (NULL == extension) {
1181: return HANDLER_GO_ON;
1182: }
1183:
1184: if (p->conf.debug) {
1185: log_error_write(srv, __FILE__, __LINE__, "s", "proxy - ext found");
1186: }
1187:
1188: if (extension->value->used == 1) {
1189: if ( ((data_proxy *)extension->value->data[0])->is_disabled ) {
1190: ndx = -1;
1191: } else {
1192: ndx = 0;
1193: }
1194: } else if (extension->value->used != 0) switch(p->conf.balance) {
1195: case PROXY_BALANCE_HASH:
1196: /* hash balancing */
1197:
1198: if (p->conf.debug) {
1199: log_error_write(srv, __FILE__, __LINE__, "sd",
1200: "proxy - used hash balancing, hosts:", extension->value->used);
1201: }
1202:
1203: for (k = 0, ndx = -1, last_max = ULONG_MAX; k < extension->value->used; k++) {
1204: data_proxy *host = (data_proxy *)extension->value->data[k];
1205: unsigned long cur_max;
1206:
1207: if (host->is_disabled) continue;
1208:
1209: cur_max = generate_crc32c(CONST_BUF_LEN(con->uri.path)) +
1210: generate_crc32c(CONST_BUF_LEN(host->host)) + /* we can cache this */
1211: generate_crc32c(CONST_BUF_LEN(con->uri.authority));
1212:
1213: if (p->conf.debug) {
1214: log_error_write(srv, __FILE__, __LINE__, "sbbbd",
1215: "proxy - election:",
1216: con->uri.path,
1217: host->host,
1218: con->uri.authority,
1219: cur_max);
1220: }
1221:
1222: if ((last_max == ULONG_MAX) || /* first round */
1223: (cur_max > last_max)) {
1224: last_max = cur_max;
1225:
1226: ndx = k;
1227: }
1228: }
1229:
1230: break;
1231: case PROXY_BALANCE_FAIR:
1232: /* fair balancing */
1233: if (p->conf.debug) {
1234: log_error_write(srv, __FILE__, __LINE__, "s",
1235: "proxy - used fair balancing");
1236: }
1237:
1238: for (k = 0, ndx = -1, max_usage = INT_MAX; k < extension->value->used; k++) {
1239: data_proxy *host = (data_proxy *)extension->value->data[k];
1240:
1241: if (host->is_disabled) continue;
1242:
1243: if (host->usage < max_usage) {
1244: max_usage = host->usage;
1245:
1246: ndx = k;
1247: }
1248: }
1249:
1250: break;
1251: case PROXY_BALANCE_RR: {
1252: data_proxy *host;
1253:
1254: /* round robin */
1255: if (p->conf.debug) {
1256: log_error_write(srv, __FILE__, __LINE__, "s",
1257: "proxy - used round-robin balancing");
1258: }
1259:
1260: /* just to be sure */
1.1.1.2 ! misho 1261: force_assert(extension->value->used < INT_MAX);
1.1 misho 1262:
1263: host = (data_proxy *)extension->value->data[0];
1264:
1265: /* Use last_used_ndx from first host in list */
1266: k = host->last_used_ndx;
1267: ndx = k + 1; /* use next host after the last one */
1268: if (ndx < 0) ndx = 0;
1269:
1270: /* Search first active host after last_used_ndx */
1271: while ( ndx < (int) extension->value->used
1272: && (host = (data_proxy *)extension->value->data[ndx])->is_disabled ) ndx++;
1273:
1274: if (ndx >= (int) extension->value->used) {
1275: /* didn't found a higher id, wrap to the start */
1276: for (ndx = 0; ndx <= (int) k; ndx++) {
1277: host = (data_proxy *)extension->value->data[ndx];
1278: if (!host->is_disabled) break;
1279: }
1280:
1281: /* No active host found */
1282: if (host->is_disabled) ndx = -1;
1283: }
1284:
1285: /* Save new index for next round */
1286: ((data_proxy *)extension->value->data[0])->last_used_ndx = ndx;
1287:
1288: break;
1289: }
1290: default:
1291: break;
1292: }
1293:
1294: /* found a server */
1295: if (ndx != -1) {
1296: data_proxy *host = (data_proxy *)extension->value->data[ndx];
1297:
1298: /*
1299: * if check-local is disabled, use the uri.path handler
1300: *
1301: */
1302:
1303: /* init handler-context */
1304: handler_ctx *hctx;
1305: hctx = handler_ctx_init();
1306:
1307: hctx->path_info_offset = path_info_offset;
1308: hctx->remote_conn = con;
1309: hctx->plugin_data = p;
1310: hctx->host = host;
1311:
1312: con->plugin_ctx[p->id] = hctx;
1313:
1314: host->usage++;
1315:
1316: con->mode = p->id;
1317:
1318: if (p->conf.debug) {
1319: log_error_write(srv, __FILE__, __LINE__, "sbd",
1320: "proxy - found a host",
1321: host->host, host->port);
1322: }
1323:
1324: return HANDLER_GO_ON;
1325: } else {
1326: /* no handler found */
1327: con->http_status = 500;
1328:
1329: log_error_write(srv, __FILE__, __LINE__, "sb",
1330: "no proxy-handler found for:",
1331: fn);
1332:
1333: return HANDLER_FINISHED;
1334: }
1335: return HANDLER_GO_ON;
1336: }
1337:
1338: static handler_t mod_proxy_connection_close_callback(server *srv, connection *con, void *p_d) {
1339: plugin_data *p = p_d;
1340:
1341: proxy_connection_close(srv, con->plugin_ctx[p->id]);
1342:
1343: return HANDLER_GO_ON;
1344: }
1345:
1346: /**
1347: *
1348: * the trigger re-enables the disabled connections after the timeout is over
1349: *
1350: * */
1351:
1352: TRIGGER_FUNC(mod_proxy_trigger) {
1353: plugin_data *p = p_d;
1354:
1355: if (p->config_storage) {
1356: size_t i, n, k;
1357: for (i = 0; i < srv->config_context->used; i++) {
1358: plugin_config *s = p->config_storage[i];
1359:
1360: if (!s) continue;
1361:
1362: /* get the extensions for all configs */
1363:
1364: for (k = 0; k < s->extensions->used; k++) {
1365: data_array *extension = (data_array *)s->extensions->data[k];
1366:
1367: /* get all hosts */
1368: for (n = 0; n < extension->value->used; n++) {
1369: data_proxy *host = (data_proxy *)extension->value->data[n];
1370:
1371: if (!host->is_disabled ||
1372: srv->cur_ts - host->disable_ts < 5) continue;
1373:
1374: log_error_write(srv, __FILE__, __LINE__, "sbd",
1375: "proxy - re-enabled:",
1376: host->host, host->port);
1377:
1378: host->is_disabled = 0;
1379: }
1380: }
1381: }
1382: }
1383:
1384: return HANDLER_GO_ON;
1385: }
1386:
1387:
1388: int mod_proxy_plugin_init(plugin *p);
1389: int mod_proxy_plugin_init(plugin *p) {
1390: p->version = LIGHTTPD_VERSION_ID;
1391: p->name = buffer_init_string("proxy");
1392:
1393: p->init = mod_proxy_init;
1394: p->cleanup = mod_proxy_free;
1395: p->set_defaults = mod_proxy_set_defaults;
1396: p->connection_reset = mod_proxy_connection_close_callback; /* end of req-resp cycle */
1397: p->handle_connection_close = mod_proxy_connection_close_callback; /* end of client connection */
1398: p->handle_uri_clean = mod_proxy_check_extension;
1399: p->handle_subrequest = mod_proxy_handle_subrequest;
1400: p->handle_trigger = mod_proxy_trigger;
1401:
1402: p->data = NULL;
1403:
1404: return 0;
1405: }