1 /*
2 * Mod Defender for HAProxy
3 *
4 * Copyright 2017 HAProxy Technologies, Dragan Dosen <ddosen@haproxy.com>
5 *
6 * Based on "A Random IP reputation service acting as a Stream Processing Offload Agent"
7 * Copyright 2016 HAProxy Technologies, Christopher Faulet <cfaulet@haproxy.com>
8 *
9 * This program is free software; you can redistribute it and/or
10 * modify it under the terms of the GNU General Public License
11 * as published by the Free Software Foundation; either version
12 * 3 of the License, or (at your option) any later version.
13 *
14 */
15 #include <unistd.h>
16 #include <stdlib.h>
17 #include <string.h>
18 #include <stdbool.h>
19 #include <errno.h>
20 #include <stdio.h>
21 #include <signal.h>
22 #include <netinet/in.h>
23 #include <sys/socket.h>
24 #include <err.h>
25 #include <ctype.h>
26
27 #include <pthread.h>
28
29 #include <event2/util.h>
30 #include <event2/event.h>
31 #include <event2/event_struct.h>
32 #include <event2/thread.h>
33
34 #include <common/mini-clist.h>
35 #include <common/chunk.h>
36
37 #include <proto/spoe.h>
38
39 #include "spoa.h"
40 #include "defender.h"
41
42 #define DEFAULT_PORT 12345
43 #define CONNECTION_BACKLOG 10
44 #define NUM_WORKERS 10
45 #define MAX_FRAME_SIZE 16384
46 #define SPOP_VERSION "2.0"
47
48 #define SLEN(str) (sizeof(str)-1)
49
50 #define DEBUG(x...) \
51 do { \
52 if (debug) \
53 LOG(x); \
54 } while (0)
55
56
57 enum spoa_state {
58 SPOA_ST_CONNECTING = 0,
59 SPOA_ST_PROCESSING,
60 SPOA_ST_DISCONNECTING,
61 };
62
63 enum spoa_frame_type {
64 SPOA_FRM_T_UNKNOWN = 0,
65 SPOA_FRM_T_HAPROXY,
66 SPOA_FRM_T_AGENT,
67 };
68
69 struct spoe_engine {
70 char *id;
71
72 struct list processing_frames;
73 struct list outgoing_frames;
74
75 struct list clients;
76 struct list list;
77 };
78
79 struct spoe_frame {
80 enum spoa_frame_type type;
81 char *buf;
82 unsigned int offset;
83 unsigned int len;
84
85 unsigned int stream_id;
86 unsigned int frame_id;
87 unsigned int flags;
88 bool hcheck; /* true is the CONNECT frame is a healthcheck */
89 bool fragmented; /* true if the frame is fragmented */
90 int defender_status; /* mod_defender returned status */
91
92 struct event process_frame_event;
93 struct worker *worker;
94 struct spoe_engine *engine;
95 struct client *client;
96 struct list list;
97
98 char *frag_buf; /* used to accumulate payload of a fragmented frame */
99 unsigned int frag_len;
100
101 char data[0];
102 };
103
104 struct client {
105 int fd;
106 unsigned long id;
107 enum spoa_state state;
108
109 struct event read_frame_event;
110 struct event write_frame_event;
111
112 struct spoe_frame *incoming_frame;
113 struct spoe_frame *outgoing_frame;
114
115 struct list processing_frames;
116 struct list outgoing_frames;
117
118 unsigned int max_frame_size;
119 int status_code;
120
121 char *engine_id;
122 struct spoe_engine *engine;
123 bool pipelining;
124 bool async;
125 bool fragmentation;
126
127 struct worker *worker;
128 struct list by_worker;
129 struct list by_engine;
130 };
131
132 /* Globals */
133 static struct worker *workers = NULL;
134 struct worker null_worker = { .id = 0 };
135 static unsigned long clicount = 0;
136 static int server_port = DEFAULT_PORT;
137 static int num_workers = NUM_WORKERS;
138 static unsigned int max_frame_size = MAX_FRAME_SIZE;
139 struct timeval processing_delay = {0, 0};
140 static bool debug = false;
141 static bool pipelining = false;
142 static bool async = false;
143 static bool fragmentation = false;
144
145
146 static const char *spoe_frm_err_reasons[SPOE_FRM_ERRS] = {
147 [SPOE_FRM_ERR_NONE] = "normal",
148 [SPOE_FRM_ERR_IO] = "I/O error",
149 [SPOE_FRM_ERR_TOUT] = "a timeout occurred",
150 [SPOE_FRM_ERR_TOO_BIG] = "frame is too big",
151 [SPOE_FRM_ERR_INVALID] = "invalid frame received",
152 [SPOE_FRM_ERR_NO_VSN] = "version value not found",
153 [SPOE_FRM_ERR_NO_FRAME_SIZE] = "max-frame-size value not found",
154 [SPOE_FRM_ERR_NO_CAP] = "capabilities value not found",
155 [SPOE_FRM_ERR_BAD_VSN] = "unsupported version",
156 [SPOE_FRM_ERR_BAD_FRAME_SIZE] = "max-frame-size too big or too small",
157 [SPOE_FRM_ERR_FRAG_NOT_SUPPORTED] = "fragmentation not supported",
158 [SPOE_FRM_ERR_INTERLACED_FRAMES] = "invalid interlaced frames",
159 [SPOE_FRM_ERR_FRAMEID_NOTFOUND] = "frame-id not found",
160 [SPOE_FRM_ERR_RES] = "resource allocation error",
161 [SPOE_FRM_ERR_UNKNOWN] = "an unknown error occurred",
162 };
163
164 static void signal_cb(evutil_socket_t, short, void *);
165 static void accept_cb(evutil_socket_t, short, void *);
166 static void worker_monitor_cb(evutil_socket_t, short, void *);
167 static void process_frame_cb(evutil_socket_t, short, void *);
168 static void read_frame_cb(evutil_socket_t, short, void *);
169 static void write_frame_cb(evutil_socket_t, short, void *);
170
171 static void use_spoe_engine(struct client *);
172 static void unuse_spoe_engine(struct client *);
173 static void release_frame(struct spoe_frame *);
174 static void release_client(struct client *);
175
176 /* Check the protocol version. It returns -1 if an error occurred, the number of
177 * read bytes otherwise. */
178 static int
check_proto_version(struct spoe_frame * frame,char ** buf,char * end)179 check_proto_version(struct spoe_frame *frame, char **buf, char *end)
180 {
181 char *str, *p = *buf;
182 uint64_t sz;
183 int ret;
184
185 /* Get the list of all supported versions by HAProxy */
186 if ((*p++ & SPOE_DATA_T_MASK) != SPOE_DATA_T_STR)
187 return -1;
188 ret = spoe_decode_buffer(&p, end, &str, &sz);
189 if (ret == -1 || !str)
190 return -1;
191
192 DEBUG(frame->worker, "<%lu> Supported versions : %.*s",
193 frame->client->id, (int)sz, str);
194
195 /* TODO: Find the right verion in supported ones */
196
197 ret = (p - *buf);
198 *buf = p;
199 return ret;
200 }
201
202 /* Check max frame size value. It returns -1 if an error occurred, the number of
203 * read bytes otherwise. */
204 static int
check_max_frame_size(struct spoe_frame * frame,char ** buf,char * end)205 check_max_frame_size(struct spoe_frame *frame, char **buf, char *end)
206 {
207 char *p = *buf;
208 uint64_t sz;
209 int type, ret;
210
211 /* Get the max-frame-size value of HAProxy */
212 type = *p++;
213 if ((type & SPOE_DATA_T_MASK) != SPOE_DATA_T_INT32 &&
214 (type & SPOE_DATA_T_MASK) != SPOE_DATA_T_INT64 &&
215 (type & SPOE_DATA_T_MASK) != SPOE_DATA_T_UINT32 &&
216 (type & SPOE_DATA_T_MASK) != SPOE_DATA_T_UINT64)
217 return -1;
218 if (decode_varint(&p, end, &sz) == -1)
219 return -1;
220
221 /* Keep the lower value */
222 if (sz < frame->client->max_frame_size)
223 frame->client->max_frame_size = sz;
224
225 DEBUG(frame->worker, "<%lu> HAProxy maximum frame size : %u",
226 frame->client->id, (unsigned int)sz);
227
228 ret = (p - *buf);
229 *buf = p;
230 return ret;
231 }
232
233 /* Check healthcheck value. It returns -1 if an error occurred, the number of
234 * read bytes otherwise. */
235 static int
check_healthcheck(struct spoe_frame * frame,char ** buf,char * end)236 check_healthcheck(struct spoe_frame *frame, char **buf, char *end)
237 {
238 char *p = *buf;
239 int type, ret;
240
241 /* Get the "healthcheck" value */
242 type = *p++;
243 if ((type & SPOE_DATA_T_MASK) != SPOE_DATA_T_BOOL)
244 return -1;
245 frame->hcheck = ((type & SPOE_DATA_FL_TRUE) == SPOE_DATA_FL_TRUE);
246
247 DEBUG(frame->worker, "<%lu> HELLO healthcheck : %s",
248 frame->client->id, (frame->hcheck ? "true" : "false"));
249
250 ret = (p - *buf);
251 *buf = p;
252 return ret;
253 }
254
255 /* Check capabilities value. It returns -1 if an error occurred, the number of
256 * read bytes otherwise. */
257 static int
check_capabilities(struct spoe_frame * frame,char ** buf,char * end)258 check_capabilities(struct spoe_frame *frame, char **buf, char *end)
259 {
260 struct client *client = frame->client;
261 char *str, *p = *buf;
262 uint64_t sz;
263 int ret;
264
265 if ((*p++ & SPOE_DATA_T_MASK) != SPOE_DATA_T_STR)
266 return -1;
267 if (spoe_decode_buffer(&p, end, &str, &sz) == -1)
268 return -1;
269 if (str == NULL) /* this is not an error */
270 goto end;
271
272 DEBUG(frame->worker, "<%lu> HAProxy capabilities : %.*s",
273 client->id, (int)sz, str);
274
275 while (sz) {
276 char *delim;
277
278 /* Skip leading spaces */
279 for (; isspace(*str) && sz; sz--);
280
281 if (sz >= 10 && !strncmp(str, "pipelining", 10)) {
282 str += 10; sz -= 10;
283 if (!sz || isspace(*str) || *str == ',') {
284 DEBUG(frame->worker,
285 "<%lu> HAProxy supports frame pipelining",
286 client->id);
287 client->pipelining = true;
288 }
289 }
290 else if (sz >= 5 && !strncmp(str, "async", 5)) {
291 str += 5; sz -= 5;
292 if (!sz || isspace(*str) || *str == ',') {
293 DEBUG(frame->worker,
294 "<%lu> HAProxy supports asynchronous frame",
295 client->id);
296 client->async = true;
297 }
298 }
299 else if (sz >= 13 && !strncmp(str, "fragmentation", 13)) {
300 str += 13; sz -= 13;
301 if (!sz || isspace(*str) || *str == ',') {
302 DEBUG(frame->worker,
303 "<%lu> HAProxy supports fragmented frame",
304 client->id);
305 client->fragmentation = true;
306 }
307 }
308
309 if (!sz || (delim = memchr(str, ',', sz)) == NULL)
310 break;
311 delim++;
312 sz -= (delim - str);
313 str = delim;
314 }
315 end:
316 ret = (p - *buf);
317 *buf = p;
318 return ret;
319 }
320
321 /* Check engine-id value. It returns -1 if an error occurred, the number of
322 * read bytes otherwise. */
323 static int
check_engine_id(struct spoe_frame * frame,char ** buf,char * end)324 check_engine_id(struct spoe_frame *frame, char **buf, char *end)
325 {
326 struct client *client = frame->client;
327 char *str, *p = *buf;
328 uint64_t sz;
329 int ret;
330
331 if ((*p++ & SPOE_DATA_T_MASK) != SPOE_DATA_T_STR)
332 return -1;
333
334 if (spoe_decode_buffer(&p, end, &str, &sz) == -1)
335 return -1;
336 if (str == NULL) /* this is not an error */
337 goto end;
338
339 if (client->engine != NULL)
340 goto end;
341
342 DEBUG(frame->worker, "<%lu> HAProxy engine id : %.*s",
343 client->id, (int)sz, str);
344
345 client->engine_id = strndup(str, (int)sz);
346 end:
347 ret = (p - *buf);
348 *buf = p;
349 return ret;
350 }
351
352 static int
acc_payload(struct spoe_frame * frame)353 acc_payload(struct spoe_frame *frame)
354 {
355 struct client *client = frame->client;
356 char *buf;
357 size_t len = frame->len - frame->offset;
358 int ret = frame->offset;
359
360 /* No need to accumulation payload */
361 if (frame->fragmented == false)
362 return ret;
363
364 buf = realloc(frame->frag_buf, frame->frag_len + len);
365 if (buf == NULL) {
366 client->status_code = SPOE_FRM_ERR_RES;
367 return -1;
368 }
369 memcpy(buf + frame->frag_len, frame->buf + frame->offset, len);
370 frame->frag_buf = buf;
371 frame->frag_len += len;
372
373 if (!(frame->flags & SPOE_FRM_FL_FIN)) {
374 /* Wait for next parts */
375 frame->buf = (char *)(frame->data);
376 frame->offset = 0;
377 frame->len = 0;
378 frame->flags = 0;
379 return 1;
380 }
381
382 frame->buf = frame->frag_buf;
383 frame->len = frame->frag_len;
384 frame->offset = 0;
385 return ret;
386 }
387
388 /* Check disconnect status code. It returns -1 if an error occurred, the number
389 * of read bytes otherwise. */
390 static int
check_discon_status_code(struct spoe_frame * frame,char ** buf,char * end)391 check_discon_status_code(struct spoe_frame *frame, char **buf, char *end)
392 {
393 char *p = *buf;
394 uint64_t sz;
395 int type, ret;
396
397 /* Get the "status-code" value */
398 type = *p++;
399 if ((type & SPOE_DATA_T_MASK) != SPOE_DATA_T_INT32 &&
400 (type & SPOE_DATA_T_MASK) != SPOE_DATA_T_INT64 &&
401 (type & SPOE_DATA_T_MASK) != SPOE_DATA_T_UINT32 &&
402 (type & SPOE_DATA_T_MASK) != SPOE_DATA_T_UINT64)
403 return -1;
404 if (decode_varint(&p, end, &sz) == -1)
405 return -1;
406
407 frame->client->status_code = (unsigned int)sz;
408
409 DEBUG(frame->worker, "<%lu> Disconnect status code : %u",
410 frame->client->id, frame->client->status_code);
411
412 ret = (p - *buf);
413 *buf = p;
414 return ret;
415 }
416
417 /* Check the disconnect message. It returns -1 if an error occurred, the number
418 * of read bytes otherwise. */
419 static int
check_discon_message(struct spoe_frame * frame,char ** buf,char * end)420 check_discon_message(struct spoe_frame *frame, char **buf, char *end)
421 {
422 char *str, *p = *buf;
423 uint64_t sz;
424 int ret;
425
426 /* Get the "message" value */
427 if ((*p++ & SPOE_DATA_T_MASK) != SPOE_DATA_T_STR)
428 return -1;
429 ret = spoe_decode_buffer(&p, end, &str, &sz);
430 if (ret == -1 || !str)
431 return -1;
432
433 DEBUG(frame->worker, "<%lu> Disconnect message : %.*s",
434 frame->client->id, (int)sz, str);
435
436 ret = (p - *buf);
437 *buf = p;
438 return ret;
439 }
440
441
442
443 /* Decode a HELLO frame received from HAProxy. It returns -1 if an error
444 * occurred, otherwise the number of read bytes. HELLO frame cannot be
445 * ignored and having another frame than a HELLO frame is an error. */
446 static int
handle_hahello(struct spoe_frame * frame)447 handle_hahello(struct spoe_frame *frame)
448 {
449 struct client *client = frame->client;
450 char *p, *end;
451
452 p = frame->buf;
453 end = frame->buf + frame->len;
454
455 /* Check frame type: we really want a HELLO frame */
456 if (*p++ != SPOE_FRM_T_HAPROXY_HELLO)
457 goto error;
458
459 DEBUG(frame->worker, "<%lu> Decode HAProxy HELLO frame", client->id);
460
461 /* Retrieve flags */
462 memcpy((char *)&(frame->flags), p, 4);
463 frame->flags = ntohl(frame->flags);
464 p += 4;
465
466 /* Fragmentation is not supported for HELLO frame */
467 if (!(frame->flags & SPOE_FRM_FL_FIN)) {
468 client->status_code = SPOE_FRM_ERR_FRAG_NOT_SUPPORTED;
469 goto error;
470 }
471
472 /* stream-id and frame-id must be cleared */
473 if (*p != 0 || *(p+1) != 0) {
474 client->status_code = SPOE_FRM_ERR_INVALID;
475 goto error;
476 }
477 p += 2;
478
479 /* Loop on K/V items */
480 while (p < end) {
481 char *str;
482 uint64_t sz;
483
484 /* Decode the item name */
485 spoe_decode_buffer(&p, end, &str, &sz);
486 if (!str) {
487 client->status_code = SPOE_FRM_ERR_INVALID;
488 goto error;
489 }
490
491 /* Check "supported-versions" K/V item */
492 if (!memcmp(str, "supported-versions", sz)) {
493 if (check_proto_version(frame, &p, end) == -1) {
494 client->status_code = SPOE_FRM_ERR_INVALID;
495 goto error;
496 }
497 }
498 /* Check "max-frame-size" K/V item */
499 else if (!memcmp(str, "max-frame-size", sz)) {
500 if (check_max_frame_size(frame, &p, end) == -1) {
501 client->status_code = SPOE_FRM_ERR_INVALID;
502 goto error;
503 }
504 }
505 /* Check "healthcheck" K/V item */
506 else if (!memcmp(str, "healthcheck", sz)) {
507 if (check_healthcheck(frame, &p, end) == -1) {
508 client->status_code = SPOE_FRM_ERR_INVALID;
509 goto error;
510 }
511 }
512 /* Check "capabilities" K/V item */
513 else if (!memcmp(str, "capabilities", sz)) {
514 if (check_capabilities(frame, &p, end) == -1) {
515 client->status_code = SPOE_FRM_ERR_INVALID;
516 goto error;
517 }
518 }
519 /* Check "engine-id" K/V item */
520 else if (!memcmp(str, "engine-id", sz)) {
521 if (check_engine_id(frame, &p, end) == -1) {
522 client->status_code = SPOE_FRM_ERR_INVALID;
523 goto error;
524 }
525 }
526 else {
527 DEBUG(frame->worker, "<%lu> Skip K/V item : key=%.*s",
528 client->id, (int)sz, str);
529
530 /* Silently ignore unknown item */
531 if (spoe_skip_data(&p, end) == -1) {
532 client->status_code = SPOE_FRM_ERR_INVALID;
533 goto error;
534 }
535 }
536 }
537
538 if (async == false || client->engine_id == NULL)
539 client->async = false;
540 if (pipelining == false)
541 client->pipelining = false;
542
543 if (client->async == true)
544 use_spoe_engine(client);
545
546 return (p - frame->buf);
547 error:
548 return -1;
549 }
550
551 /* Decode a DISCONNECT frame received from HAProxy. It returns -1 if an error
552 * occurred, otherwise the number of read bytes. DISCONNECT frame cannot be
553 * ignored and having another frame than a DISCONNECT frame is an error.*/
554 static int
handle_hadiscon(struct spoe_frame * frame)555 handle_hadiscon(struct spoe_frame *frame)
556 {
557 struct client *client = frame->client;
558 char *p, *end;
559
560 p = frame->buf;
561 end = frame->buf + frame->len;
562
563 /* Check frame type: we really want a DISCONNECT frame */
564 if (*p++ != SPOE_FRM_T_HAPROXY_DISCON)
565 goto error;
566
567 DEBUG(frame->worker, "<%lu> Decode HAProxy DISCONNECT frame", client->id);
568
569 /* Retrieve flags */
570 memcpy((char *)&(frame->flags), p, 4);
571 frame->flags = ntohl(frame->flags);
572 p += 4;
573
574 /* Fragmentation is not supported for DISCONNECT frame */
575 if (!(frame->flags & SPOE_FRM_FL_FIN)) {
576 client->status_code = SPOE_FRM_ERR_FRAG_NOT_SUPPORTED;
577 goto error;
578 }
579
580 /* stream-id and frame-id must be cleared */
581 if (*p != 0 || *(p+1) != 0) {
582 client->status_code = SPOE_FRM_ERR_INVALID;
583 goto error;
584 }
585 p += 2;
586
587 client->status_code = SPOE_FRM_ERR_NONE;
588
589 /* Loop on K/V items */
590 while (p < end) {
591 char *str;
592 uint64_t sz;
593
594 /* Decode item key */
595 spoe_decode_buffer(&p, end, &str, &sz);
596 if (!str) {
597 client->status_code = SPOE_FRM_ERR_INVALID;
598 goto error;
599 }
600
601 /* Check "status-code" K/V item */
602 if (!memcmp(str, "status-code", sz)) {
603 if (check_discon_status_code(frame, &p, end) == -1) {
604 client->status_code = SPOE_FRM_ERR_INVALID;
605 goto error;
606 }
607 }
608 /* Check "message" K/V item */
609 else if (!memcmp(str, "message", sz)) {
610 if (check_discon_message(frame, &p, end) == -1) {
611 client->status_code = SPOE_FRM_ERR_INVALID;
612 goto error;
613 }
614 }
615 else {
616 DEBUG(frame->worker, "<%lu> Skip K/V item : key=%.*s",
617 client->id, (int)sz, str);
618
619 /* Silently ignore unknown item */
620 if (spoe_skip_data(&p, end) == -1) {
621 client->status_code = SPOE_FRM_ERR_INVALID;
622 goto error;
623 }
624 }
625 }
626
627 return (p - frame->buf);
628 error:
629 return -1;
630 }
631
632 /* Decode a NOTIFY frame received from HAProxy. It returns -1 if an error
633 * occurred, 0 if it must be must be ignored, otherwise the number of read
634 * bytes. */
635 static int
handle_hanotify(struct spoe_frame * frame)636 handle_hanotify(struct spoe_frame *frame)
637 {
638 struct client *client = frame->client;
639 char *p, *end;
640 uint64_t stream_id, frame_id;
641
642 p = frame->buf;
643 end = frame->buf + frame->len;
644
645 /* Check frame type */
646 if (*p++ != SPOE_FRM_T_HAPROXY_NOTIFY)
647 goto ignore;
648
649 DEBUG(frame->worker, "<%lu> Decode HAProxy NOTIFY frame", client->id);
650
651 /* Retrieve flags */
652 memcpy((char *)&(frame->flags), p, 4);
653 frame->flags = ntohl(frame->flags);
654 p += 4;
655
656 /* Fragmentation is not supported */
657 if (!(frame->flags & SPOE_FRM_FL_FIN) && fragmentation == false) {
658 client->status_code = SPOE_FRM_ERR_FRAG_NOT_SUPPORTED;
659 goto error;
660 }
661
662 /* Read the stream-id and frame-id */
663 if (decode_varint(&p, end, &stream_id) == -1)
664 goto ignore;
665 if (decode_varint(&p, end, &frame_id) == -1)
666 goto ignore;
667
668 frame->stream_id = (unsigned int)stream_id;
669 frame->frame_id = (unsigned int)frame_id;
670
671 DEBUG(frame->worker, "<%lu> STREAM-ID=%u - FRAME-ID=%u"
672 " - %s frame received"
673 " - frag_len=%u - len=%u - offset=%ld",
674 client->id, frame->stream_id, frame->frame_id,
675 (frame->flags & SPOE_FRM_FL_FIN) ? "unfragmented" : "fragmented",
676 frame->frag_len, frame->len, p - frame->buf);
677
678 frame->fragmented = !(frame->flags & SPOE_FRM_FL_FIN);
679 frame->offset = (p - frame->buf);
680 return acc_payload(frame);
681
682 ignore:
683 return 0;
684
685 error:
686 return -1;
687 }
688
689 /* Decode next part of a fragmented frame received from HAProxy. It returns -1
690 * if an error occurred, 0 if it must be must be ignored, otherwise the number
691 * of read bytes. */
692 static int
handle_hafrag(struct spoe_frame * frame)693 handle_hafrag(struct spoe_frame *frame)
694 {
695 struct client *client = frame->client;
696 char *p, *end;
697 uint64_t stream_id, frame_id;
698
699 p = frame->buf;
700 end = frame->buf + frame->len;
701
702 /* Check frame type */
703 if (*p++ != SPOE_FRM_T_UNSET)
704 goto ignore;
705
706 DEBUG(frame->worker, "<%lu> Decode Next part of a fragmented frame", client->id);
707
708 /* Fragmentation is not supported */
709 if (fragmentation == false) {
710 client->status_code = SPOE_FRM_ERR_FRAG_NOT_SUPPORTED;
711 goto error;
712 }
713
714 /* Retrieve flags */
715 memcpy((char *)&(frame->flags), p, 4);
716 frame->flags = ntohl(frame->flags);
717 p+= 4;
718
719 /* Read the stream-id and frame-id */
720 if (decode_varint(&p, end, &stream_id) == -1)
721 goto ignore;
722 if (decode_varint(&p, end, &frame_id) == -1)
723 goto ignore;
724
725 if (frame->fragmented == false ||
726 frame->stream_id != (unsigned int)stream_id ||
727 frame->frame_id != (unsigned int)frame_id) {
728 client->status_code = SPOE_FRM_ERR_INTERLACED_FRAMES;
729 goto error;
730 }
731
732 if (frame->flags & SPOE_FRM_FL_ABRT) {
733 DEBUG(frame->worker, "<%lu> STREAM-ID=%u - FRAME-ID=%u"
734 " - Abort processing of a fragmented frame"
735 " - frag_len=%u - len=%u - offset=%ld",
736 client->id, frame->stream_id, frame->frame_id,
737 frame->frag_len, frame->len, p - frame->buf);
738 goto ignore;
739 }
740
741 DEBUG(frame->worker, "<%lu> STREAM-ID=%u - FRAME-ID=%u"
742 " - %s fragment of a fragmented frame received"
743 " - frag_len=%u - len=%u - offset=%ld",
744 client->id, frame->stream_id, frame->frame_id,
745 (frame->flags & SPOE_FRM_FL_FIN) ? "last" : "next",
746 frame->frag_len, frame->len, p - frame->buf);
747
748 frame->offset = (p - frame->buf);
749 return acc_payload(frame);
750
751 ignore:
752 return 0;
753
754 error:
755 return -1;
756 }
757
758 /* Encode a HELLO frame to send it to HAProxy. It returns the number of written
759 * bytes. */
760 static int
prepare_agenthello(struct spoe_frame * frame)761 prepare_agenthello(struct spoe_frame *frame)
762 {
763 struct client *client = frame->client;
764 char *p, *end;
765 char capabilities[64];
766 int n;
767 unsigned int flags = SPOE_FRM_FL_FIN;
768
769 DEBUG(frame->worker, "<%lu> Encode Agent HELLO frame", client->id);
770 frame->type = SPOA_FRM_T_AGENT;
771
772 p = frame->buf;
773 end = frame->buf+max_frame_size;
774
775 /* Frame Type */
776 *p++ = SPOE_FRM_T_AGENT_HELLO;
777
778 /* Set flags */
779 flags = htonl(flags);
780 memcpy(p, (char *)&flags, 4);
781 p += 4;
782
783 /* No stream-id and frame-id for HELLO frames */
784 *p++ = 0;
785 *p++ = 0;
786
787 /* "version" K/V item */
788 spoe_encode_buffer("version", 7, &p, end);
789 *p++ = SPOE_DATA_T_STR;
790 spoe_encode_buffer(SPOP_VERSION, SLEN(SPOP_VERSION), &p, end);
791 DEBUG(frame->worker, "<%lu> Agent version : %s",
792 client->id, SPOP_VERSION);
793
794
795 /* "max-frame-size" K/V item */
796 spoe_encode_buffer("max-frame-size", 14, &p ,end);
797 *p++ = SPOE_DATA_T_UINT32;
798 encode_varint(client->max_frame_size, &p, end);
799 DEBUG(frame->worker, "<%lu> Agent maximum frame size : %u",
800 client->id, client->max_frame_size);
801
802 /* "capabilities" K/V item */
803 spoe_encode_buffer("capabilities", 12, &p, end);
804 *p++ = SPOE_DATA_T_STR;
805
806 memset(capabilities, 0, sizeof(capabilities));
807 n = 0;
808
809 /* 1. Fragmentation capability ? */
810 if (fragmentation == true) {
811 memcpy(capabilities, "fragmentation", 13);
812 n += 13;
813 }
814 /* 2. Pipelining capability ? */
815 if (client->pipelining == true) {
816 if (n) capabilities[n++] = ',';
817 memcpy(capabilities + n, "pipelining", 10);
818 n += 10;
819 }
820 /* 3. Async capability ? */
821 if (client->async == true) {
822 if (n) capabilities[n++] = ',';
823 memcpy(capabilities + n, "async", 5);
824 n += 5;
825 }
826 spoe_encode_buffer(capabilities, n, &p, end);
827
828 DEBUG(frame->worker, "<%lu> Agent capabilities : %.*s",
829 client->id, n, capabilities);
830
831 frame->len = (p - frame->buf);
832 return frame->len;
833 }
834
835 /* Encode a DISCONNECT frame to send it to HAProxy. It returns the number of
836 * written bytes. */
837 static int
prepare_agentdicon(struct spoe_frame * frame)838 prepare_agentdicon(struct spoe_frame *frame)
839 {
840 struct client *client = frame->client;
841 char *p, *end;
842 const char *reason;
843 int rlen;
844 unsigned int flags = SPOE_FRM_FL_FIN;
845
846 DEBUG(frame->worker, "<%lu> Encode Agent DISCONNECT frame", client->id);
847 frame->type = SPOA_FRM_T_AGENT;
848
849 p = frame->buf;
850 end = frame->buf+max_frame_size;
851
852 if (client->status_code >= SPOE_FRM_ERRS)
853 client->status_code = SPOE_FRM_ERR_UNKNOWN;
854 reason = spoe_frm_err_reasons[client->status_code];
855 rlen = strlen(reason);
856
857 /* Frame type */
858 *p++ = SPOE_FRM_T_AGENT_DISCON;
859
860 /* Set flags */
861 flags = htonl(flags);
862 memcpy(p, (char *)&flags, 4);
863 p += 4;
864
865 /* No stream-id and frame-id for DISCONNECT frames */
866 *p++ = 0;
867 *p++ = 0;
868
869 /* There are 2 mandatory items: "status-code" and "message" */
870
871 /* "status-code" K/V item */
872 spoe_encode_buffer("status-code", 11, &p, end);
873 *p++ = SPOE_DATA_T_UINT32;
874 encode_varint(client->status_code, &p, end);
875 DEBUG(frame->worker, "<%lu> Disconnect status code : %u",
876 client->id, client->status_code);
877
878 /* "message" K/V item */
879 spoe_encode_buffer("message", 7, &p, end);
880 *p++ = SPOE_DATA_T_STR;
881 spoe_encode_buffer(reason, rlen, &p, end);
882 DEBUG(frame->worker, "<%lu> Disconnect message : %s",
883 client->id, reason);
884
885 frame->len = (p - frame->buf);
886 return frame->len;
887 }
888
889 /* Encode a ACK frame to send it to HAProxy. It returns the number of written
890 * bytes. */
891 static int
prepare_agentack(struct spoe_frame * frame)892 prepare_agentack(struct spoe_frame *frame)
893 {
894 char *p, *end;
895 unsigned int flags = SPOE_FRM_FL_FIN;
896
897 /* Be careful here, in async mode, frame->client can be NULL */
898
899 DEBUG(frame->worker, "Encode Agent ACK frame");
900 frame->type = SPOA_FRM_T_AGENT;
901
902 p = frame->buf;
903 end = frame->buf+max_frame_size;
904
905 /* Frame type */
906 *p++ = SPOE_FRM_T_AGENT_ACK;
907
908 /* Set flags */
909 flags = htonl(flags);
910 memcpy(p, (char *)&flags, 4);
911 p += 4;
912
913 /* Set stream-id and frame-id for ACK frames */
914 encode_varint(frame->stream_id, &p, end);
915 encode_varint(frame->frame_id, &p, end);
916
917 DEBUG(frame->worker, "STREAM-ID=%u - FRAME-ID=%u",
918 frame->stream_id, frame->frame_id);
919
920 frame->len = (p - frame->buf);
921 return frame->len;
922 }
923
924 static int
create_server_socket(void)925 create_server_socket(void)
926 {
927 struct sockaddr_in listen_addr;
928 int fd, yes = 1;
929
930 fd = socket(AF_INET, SOCK_STREAM, 0);
931 if (fd < 0) {
932 LOG(&null_worker, "Failed to create service socket : %m");
933 return -1;
934 }
935
936 memset(&listen_addr, 0, sizeof(listen_addr));
937 listen_addr.sin_family = AF_INET;
938 listen_addr.sin_addr.s_addr = INADDR_ANY;
939 listen_addr.sin_port = htons(server_port);
940
941 if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(yes)) < 0 ||
942 setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &yes, sizeof(yes)) < 0) {
943 LOG(&null_worker, "Failed to set option on server socket : %m");
944 return -1;
945 }
946
947 if (bind(fd, (struct sockaddr *) &listen_addr, sizeof(listen_addr)) < 0) {
948 LOG(&null_worker, "Failed to bind server socket : %m");
949 return -1;
950 }
951
952 if (listen(fd, CONNECTION_BACKLOG) < 0) {
953 LOG(&null_worker, "Failed to listen on server socket : %m");
954 return -1;
955 }
956
957 return fd;
958 }
959
960 static void
release_frame(struct spoe_frame * frame)961 release_frame(struct spoe_frame *frame)
962 {
963 struct worker *worker;
964
965 if (frame == NULL)
966 return;
967
968 if (event_pending(&frame->process_frame_event, EV_TIMEOUT, NULL))
969 event_del(&frame->process_frame_event);
970
971 worker = frame->worker;
972 LIST_DEL(&frame->list);
973 if (frame->frag_buf)
974 free(frame->frag_buf);
975 memset(frame, 0, sizeof(*frame)+max_frame_size+4);
976 LIST_ADDQ(&worker->frames, &frame->list);
977 }
978
979 static void
release_client(struct client * c)980 release_client(struct client *c)
981 {
982 struct spoe_frame *frame, *back;
983
984 if (c == NULL)
985 return;
986
987 DEBUG(c->worker, "<%lu> Release client", c->id);
988
989 LIST_DEL(&c->by_worker);
990 c->worker->nbclients--;
991
992 unuse_spoe_engine(c);
993 free(c->engine_id);
994
995 if (event_pending(&c->read_frame_event, EV_READ, NULL))
996 event_del(&c->read_frame_event);
997 if (event_pending(&c->write_frame_event, EV_WRITE, NULL))
998 event_del(&c->write_frame_event);
999
1000 release_frame(c->incoming_frame);
1001 release_frame(c->outgoing_frame);
1002 list_for_each_entry_safe(frame, back, &c->processing_frames, list) {
1003 release_frame(frame);
1004 }
1005 list_for_each_entry_safe(frame, back, &c->outgoing_frames, list) {
1006 release_frame(frame);
1007 }
1008
1009 if (c->fd >= 0)
1010 close(c->fd);
1011
1012 free(c);
1013 }
1014
1015 static void
reset_frame(struct spoe_frame * frame)1016 reset_frame(struct spoe_frame *frame)
1017 {
1018 if (frame == NULL)
1019 return;
1020
1021 if (frame->frag_buf)
1022 free(frame->frag_buf);
1023
1024 frame->type = SPOA_FRM_T_UNKNOWN;
1025 frame->buf = (char *)(frame->data);
1026 frame->offset = 0;
1027 frame->len = 0;
1028 frame->stream_id = 0;
1029 frame->frame_id = 0;
1030 frame->flags = 0;
1031 frame->hcheck = false;
1032 frame->fragmented = false;
1033 frame->defender_status = -1;
1034 frame->frag_buf = NULL;
1035 frame->frag_len = 0;
1036 LIST_INIT(&frame->list);
1037 }
1038
1039 static void
use_spoe_engine(struct client * client)1040 use_spoe_engine(struct client *client)
1041 {
1042 struct spoe_engine *eng;
1043
1044 if (client->engine_id == NULL)
1045 return;
1046
1047 list_for_each_entry(eng, &client->worker->engines, list) {
1048 if (!strcmp(eng->id, client->engine_id))
1049 goto end;
1050 }
1051
1052 if ((eng = malloc(sizeof(*eng))) == NULL) {
1053 client->async = false;
1054 return;
1055 }
1056
1057 eng->id = strdup(client->engine_id);
1058 LIST_INIT(&eng->clients);
1059 LIST_INIT(&eng->processing_frames);
1060 LIST_INIT(&eng->outgoing_frames);
1061 LIST_ADDQ(&client->worker->engines, &eng->list);
1062 LOG(client->worker, "Add new SPOE engine '%s'", eng->id);
1063
1064 end:
1065 client->engine = eng;
1066 LIST_ADDQ(&eng->clients, &client->by_engine);
1067 }
1068
1069 static void
unuse_spoe_engine(struct client * client)1070 unuse_spoe_engine(struct client *client)
1071 {
1072 struct spoe_engine *eng;
1073 struct spoe_frame *frame, *back;
1074
1075 if (client == NULL || client->engine == NULL)
1076 return;
1077
1078 eng = client->engine;
1079 client->engine = NULL;
1080 LIST_DEL(&client->by_engine);
1081 if (!LIST_ISEMPTY(&eng->clients))
1082 return;
1083
1084 LOG(client->worker, "Remove SPOE engine '%s'", eng->id);
1085 LIST_DEL(&eng->list);
1086
1087 list_for_each_entry_safe(frame, back, &eng->processing_frames, list) {
1088 release_frame(frame);
1089 }
1090 list_for_each_entry_safe(frame, back, &eng->outgoing_frames, list) {
1091 release_frame(frame);
1092 }
1093 free(eng->id);
1094 free(eng);
1095 }
1096
1097
1098 static struct spoe_frame *
acquire_incoming_frame(struct client * client)1099 acquire_incoming_frame(struct client *client)
1100 {
1101 struct spoe_frame *frame;
1102
1103 frame = client->incoming_frame;
1104 if (frame != NULL)
1105 return frame;
1106
1107 if (LIST_ISEMPTY(&client->worker->frames)) {
1108 if ((frame = calloc(1, sizeof(*frame)+max_frame_size+4)) == NULL) {
1109 LOG(client->worker, "Failed to allocate new frame : %m");
1110 return NULL;
1111 }
1112 }
1113 else {
1114 frame = LIST_NEXT(&client->worker->frames, typeof(frame), list);
1115 LIST_DEL(&frame->list);
1116 }
1117
1118 reset_frame(frame);
1119 frame->worker = client->worker;
1120 frame->engine = client->engine;
1121 frame->client = client;
1122
1123 if (event_assign(&frame->process_frame_event, client->worker->base, -1,
1124 EV_TIMEOUT|EV_PERSIST, process_frame_cb, frame) < 0) {
1125 LOG(client->worker, "Failed to create frame event");
1126 return NULL;
1127 }
1128
1129 client->incoming_frame = frame;
1130 return frame;
1131 }
1132
1133 static struct spoe_frame *
acquire_outgoing_frame(struct client * client)1134 acquire_outgoing_frame(struct client *client)
1135 {
1136 struct spoe_engine *engine = client->engine;
1137 struct spoe_frame *frame = NULL;
1138
1139 if (client->outgoing_frame != NULL)
1140 frame = client->outgoing_frame;
1141 else if (!LIST_ISEMPTY(&client->outgoing_frames)) {
1142 frame = LIST_NEXT(&client->outgoing_frames, typeof(frame), list);
1143 LIST_DEL(&frame->list);
1144 client->outgoing_frame = frame;
1145 }
1146 else if (engine!= NULL && !LIST_ISEMPTY(&engine->outgoing_frames)) {
1147 frame = LIST_NEXT(&engine->outgoing_frames, typeof(frame), list);
1148 LIST_DEL(&frame->list);
1149 client->outgoing_frame = frame;
1150 }
1151 return frame;
1152 }
1153
1154 static void
write_frame(struct client * client,struct spoe_frame * frame)1155 write_frame(struct client *client, struct spoe_frame *frame)
1156 {
1157 uint32_t netint;
1158
1159 LIST_DEL(&frame->list);
1160
1161 frame->buf = (char *)(frame->data);
1162 frame->offset = 0;
1163 netint = htonl(frame->len);
1164 memcpy(frame->buf, &netint, 4);
1165
1166 if (client != NULL) { /* HELLO or DISCONNECT frames */
1167 event_add(&client->write_frame_event, NULL);
1168
1169 /* Try to process the frame as soon as possible, and always
1170 * attach it to the client */
1171 if (client->async || client->pipelining) {
1172 if (client->outgoing_frame == NULL)
1173 client->outgoing_frame = frame;
1174 else
1175 LIST_ADD(&client->outgoing_frames, &frame->list);
1176 }
1177 else {
1178 client->outgoing_frame = frame;
1179 event_del(&client->read_frame_event);
1180 }
1181 }
1182 else { /* for all other frames */
1183 if (frame->client == NULL) { /* async mode ! */
1184 LIST_ADDQ(&frame->engine->outgoing_frames, &frame->list);
1185 list_for_each_entry(client, &frame->engine->clients, by_engine)
1186 event_add(&client->write_frame_event, NULL);
1187 }
1188 else if (frame->client->pipelining) {
1189 LIST_ADDQ(&frame->client->outgoing_frames, &frame->list);
1190 event_add(&frame->client->write_frame_event, NULL);
1191 }
1192 else {
1193 frame->client->outgoing_frame = frame;
1194 event_add(&frame->client->write_frame_event, NULL);
1195 event_del(&frame->client->read_frame_event);
1196 }
1197 }
1198 }
1199
1200 static void
process_incoming_frame(struct spoe_frame * frame)1201 process_incoming_frame(struct spoe_frame *frame)
1202 {
1203 struct client *client = frame->client;
1204
1205 if (event_add(&frame->process_frame_event, &processing_delay) < 0) {
1206 LOG(client->worker, "Failed to process incoming frame");
1207 release_frame(frame);
1208 return;
1209 }
1210
1211 if (client->async) {
1212 frame->client = NULL;
1213 LIST_ADDQ(&frame->engine->processing_frames, &frame->list);
1214 }
1215 else if (client->pipelining)
1216 LIST_ADDQ(&client->processing_frames, &frame->list);
1217 else
1218 event_del(&client->read_frame_event);
1219 }
1220
1221 static void
signal_cb(evutil_socket_t sig,short events,void * user_data)1222 signal_cb(evutil_socket_t sig, short events, void *user_data)
1223 {
1224 struct event_base *base = user_data;
1225 int i;
1226
1227 DEBUG(&null_worker, "Stopping the server");
1228
1229 event_base_loopbreak(base);
1230 DEBUG(&null_worker, "Main event loop stopped");
1231
1232 for (i = 0; i < num_workers; i++) {
1233 event_base_loopbreak(workers[i].base);
1234 DEBUG(&null_worker, "Event loop stopped for worker %02d",
1235 workers[i].id);
1236 }
1237 }
1238
1239 static void
worker_monitor_cb(evutil_socket_t fd,short events,void * arg)1240 worker_monitor_cb(evutil_socket_t fd, short events, void *arg)
1241 {
1242 struct worker *worker = arg;
1243
1244 LOG(worker, "%u clients connected", worker->nbclients);
1245 }
1246
1247 static void
process_frame_cb(evutil_socket_t fd,short events,void * arg)1248 process_frame_cb(evutil_socket_t fd, short events, void *arg)
1249 {
1250 struct spoe_frame *frame = arg;
1251 char *p, *end;
1252 int ret;
1253
1254 DEBUG(frame->worker,
1255 "Process frame messages : STREAM-ID=%u - FRAME-ID=%u - length=%u bytes",
1256 frame->stream_id, frame->frame_id, frame->len - frame->offset);
1257
1258 p = frame->buf + frame->offset;
1259 end = frame->buf + frame->len;
1260
1261 /* Loop on messages */
1262 while (p < end) {
1263 char *str;
1264 uint64_t sz;
1265 int nbargs;
1266
1267 /* Decode the message name */
1268 spoe_decode_buffer(&p, end, &str, &sz);
1269 if (!str)
1270 goto stop_processing;
1271
1272 DEBUG(frame->worker, "Process SPOE Message '%.*s'", (int)sz, str);
1273
1274 nbargs = *p++; /* Get the number of arguments */
1275 frame->offset = (p - frame->buf); /* Save index to handle errors and skip args */
1276 if (!memcmp(str, "check-request", sz)) {
1277 struct defender_request request;
1278
1279 memset(&request, 0, sizeof(request));
1280
1281 if (nbargs != 8)
1282 goto skip_message;
1283
1284 if (spoe_decode_buffer(&p, end, &str, &sz) == -1)
1285 goto stop_processing;
1286 if (spoe_decode_data(&p, end, &request.clientip) == -1)
1287 goto skip_message;
1288
1289 if (spoe_decode_buffer(&p, end, &str, &sz) == -1)
1290 goto stop_processing;
1291 if (spoe_decode_data(&p, end, &request.id) == -1)
1292 goto skip_message;
1293
1294 if (spoe_decode_buffer(&p, end, &str, &sz) == -1)
1295 goto stop_processing;
1296 if (spoe_decode_data(&p, end, &request.method) == -1)
1297 goto skip_message;
1298
1299 if (spoe_decode_buffer(&p, end, &str, &sz) == -1)
1300 goto stop_processing;
1301 if (spoe_decode_data(&p, end, &request.path) == -1)
1302 goto skip_message;
1303
1304 if (spoe_decode_buffer(&p, end, &str, &sz) == -1)
1305 goto stop_processing;
1306 if (spoe_decode_data(&p, end, &request.query) == -1)
1307 goto skip_message;
1308
1309 if (spoe_decode_buffer(&p, end, &str, &sz) == -1)
1310 goto stop_processing;
1311 if (spoe_decode_data(&p, end, &request.version) == -1)
1312 goto skip_message;
1313
1314 if (spoe_decode_buffer(&p, end, &str, &sz) == -1)
1315 goto stop_processing;
1316 if (spoe_decode_data(&p, end, &request.headers) == -1)
1317 goto skip_message;
1318
1319 if (spoe_decode_buffer(&p, end, &str, &sz) == -1)
1320 goto stop_processing;
1321 if (spoe_decode_data(&p, end, &request.body) == -1)
1322 goto skip_message;
1323
1324 frame->defender_status = defender_process_request(frame->worker, &request);
1325 }
1326 else {
1327 skip_message:
1328 p = frame->buf + frame->offset; /* Restore index */
1329
1330 while (nbargs-- > 0) {
1331 /* Silently ignore argument: its name and its value */
1332 if (spoe_decode_buffer(&p, end, &str, &sz) == -1)
1333 goto stop_processing;
1334 if (spoe_skip_data(&p, end) == -1)
1335 goto stop_processing;
1336 }
1337 }
1338 }
1339
1340 stop_processing:
1341 /* Prepare agent ACK frame */
1342 frame->buf = (char *)(frame->data) + 4;
1343 frame->offset = 0;
1344 frame->len = 0;
1345 frame->flags = 0;
1346
1347 ret = prepare_agentack(frame);
1348 p = frame->buf + ret;
1349 end = frame->buf+max_frame_size;
1350
1351 if (frame->defender_status != -1) {
1352 DEBUG(frame->worker, "Add action : set variable status=%u",
1353 frame->defender_status);
1354
1355 *p++ = SPOE_ACT_T_SET_VAR; /* Action type */
1356 *p++ = 3; /* Number of args */
1357 *p++ = SPOE_SCOPE_SESS; /* Arg 1: the scope */
1358 spoe_encode_buffer("status", 6, &p, end); /* Arg 2: variable name */
1359 *p++ = SPOE_DATA_T_UINT32;
1360 encode_varint(frame->defender_status, &p, end); /* Arg 3: variable value */
1361 frame->len = (p - frame->buf);
1362 }
1363 write_frame(NULL, frame);
1364 }
1365
1366 static void
read_frame_cb(evutil_socket_t fd,short events,void * arg)1367 read_frame_cb(evutil_socket_t fd, short events, void *arg)
1368 {
1369 struct client *client = arg;
1370 struct spoe_frame *frame;
1371 uint32_t netint;
1372 int n;
1373
1374 DEBUG(client->worker, "<%lu> %s", client->id, __FUNCTION__);
1375 if ((frame = acquire_incoming_frame(client)) == NULL)
1376 goto close;
1377
1378 frame->type = SPOA_FRM_T_HAPROXY;
1379 if (frame->buf == (char *)(frame->data)) {
1380 /* Read the frame length: frame->buf points on length part (frame->data) */
1381 n = read(client->fd, frame->buf+frame->offset, 4-frame->offset);
1382 if (n <= 0) {
1383 if (n < 0)
1384 LOG(client->worker, "Failed to read frame length : %m");
1385 goto close;
1386 }
1387 frame->offset += n;
1388 if (frame->offset != 4)
1389 return;
1390 memcpy(&netint, frame->buf, 4);
1391 frame->buf += 4;
1392 frame->offset = 0;
1393 frame->len = ntohl(netint);
1394 }
1395
1396 /* Read the frame: frame->buf points on frame part (frame->data+4)*/
1397 n = read(client->fd, frame->buf + frame->offset,
1398 frame->len - frame->offset);
1399 if (n <= 0) {
1400 if (n < 0) {
1401 LOG(client->worker, "Frame to read frame : %m");
1402 goto close;
1403 }
1404 return;
1405 }
1406 frame->offset += n;
1407 if (frame->offset != frame->len)
1408 return;
1409 frame->offset = 0;
1410
1411 DEBUG(client->worker, "<%lu> New Frame of %u bytes received",
1412 client->id, frame->len);
1413
1414 switch (client->state) {
1415 case SPOA_ST_CONNECTING:
1416 if (handle_hahello(frame) < 0) {
1417 LOG(client->worker, "Failed to decode HELLO frame");
1418 goto disconnect;
1419 }
1420 prepare_agenthello(frame);
1421 goto write_frame;
1422
1423 case SPOA_ST_PROCESSING:
1424 if (frame->buf[0] == SPOE_FRM_T_HAPROXY_DISCON) {
1425 client->state = SPOA_ST_DISCONNECTING;
1426 goto disconnecting;
1427 }
1428 if (frame->buf[0] == SPOE_FRM_T_UNSET)
1429 n = handle_hafrag(frame);
1430 else
1431 n = handle_hanotify(frame);
1432
1433 if (n < 0) {
1434 LOG(client->worker, "Failed to decode frame: %s",
1435 spoe_frm_err_reasons[client->status_code]);
1436 goto disconnect;
1437 }
1438 else if (n == 0) {
1439 LOG(client->worker, "Ignore invalid/unknown/aborted frame");
1440 goto ignore_frame;
1441 }
1442 else if (n == 1)
1443 goto noop;
1444 else
1445 goto process_frame;
1446
1447 case SPOA_ST_DISCONNECTING:
1448 disconnecting:
1449 if (handle_hadiscon(frame) < 0) {
1450 LOG(client->worker, "Failed to decode DISCONNECT frame");
1451 goto disconnect;
1452 }
1453 if (client->status_code != SPOE_FRM_ERR_NONE)
1454 LOG(client->worker, "<%lu> Peer closed connection: %s",
1455 client->id, spoe_frm_err_reasons[client->status_code]);
1456 goto disconnect;
1457 }
1458
1459 noop:
1460 return;
1461
1462 ignore_frame:
1463 reset_frame(frame);
1464 return;
1465
1466 process_frame:
1467 process_incoming_frame(frame);
1468 client->incoming_frame = NULL;
1469 return;
1470
1471 write_frame:
1472 write_frame(client, frame);
1473 client->incoming_frame = NULL;
1474 return;
1475
1476 disconnect:
1477 client->state = SPOA_ST_DISCONNECTING;
1478 if (prepare_agentdicon(frame) < 0) {
1479 LOG(client->worker, "Failed to encode DISCONNECT frame");
1480 goto close;
1481 }
1482 goto write_frame;
1483
1484 close:
1485 release_client(client);
1486 }
1487
1488 static void
write_frame_cb(evutil_socket_t fd,short events,void * arg)1489 write_frame_cb(evutil_socket_t fd, short events, void *arg)
1490 {
1491 struct client *client = arg;
1492 struct spoe_frame *frame;
1493 int n;
1494
1495 DEBUG(client->worker, "<%lu> %s", client->id, __FUNCTION__);
1496 if ((frame = acquire_outgoing_frame(client)) == NULL) {
1497 event_del(&client->write_frame_event);
1498 return;
1499 }
1500
1501 if (frame->buf == (char *)(frame->data)) {
1502 /* Write the frame length: frame->buf points on length part (frame->data) */
1503 n = write(client->fd, frame->buf+frame->offset, 4-frame->offset);
1504 if (n <= 0) {
1505 if (n < 0)
1506 LOG(client->worker, "Failed to write frame length : %m");
1507 goto close;
1508 }
1509 frame->offset += n;
1510 if (frame->offset != 4)
1511 return;
1512 frame->buf += 4;
1513 frame->offset = 0;
1514 }
1515
1516 /* Write the frame: frame->buf points on frame part (frame->data+4)*/
1517 n = write(client->fd, frame->buf + frame->offset,
1518 frame->len - frame->offset);
1519 if (n <= 0) {
1520 if (n < 0) {
1521 LOG(client->worker, "Failed to write frame : %m");
1522 goto close;
1523 }
1524 return;
1525 }
1526 frame->offset += n;
1527 if (frame->offset != frame->len)
1528 return;
1529
1530 DEBUG(client->worker, "<%lu> Frame of %u bytes send",
1531 client->id, frame->len);
1532
1533 switch (client->state) {
1534 case SPOA_ST_CONNECTING:
1535 if (frame->hcheck == true) {
1536 DEBUG(client->worker,
1537 "<%lu> Close client after healthcheck",
1538 client->id);
1539 goto close;
1540 }
1541 client->state = SPOA_ST_PROCESSING;
1542 break;
1543
1544 case SPOA_ST_PROCESSING:
1545 break;
1546
1547 case SPOA_ST_DISCONNECTING:
1548 goto close;
1549 }
1550
1551 release_frame(frame);
1552 client->outgoing_frame = NULL;
1553 if (!client->async && !client->pipelining) {
1554 event_del(&client->write_frame_event);
1555 event_add(&client->read_frame_event, NULL);
1556 }
1557 return;
1558
1559 close:
1560 release_client(client);
1561 }
1562
1563 static void
accept_cb(int listener,short event,void * arg)1564 accept_cb(int listener, short event, void *arg)
1565 {
1566 struct worker *worker;
1567 struct client *client;
1568 int fd;
1569
1570 worker = &workers[clicount++ % num_workers];
1571
1572 if ((fd = accept(listener, NULL, NULL)) < 0) {
1573 if (errno != EAGAIN && errno != EWOULDBLOCK)
1574 LOG(worker, "Failed to accept client connection : %m");
1575 return;
1576 }
1577
1578 DEBUG(&null_worker,
1579 "<%lu> New Client connection accepted and assigned to worker %02d",
1580 clicount, worker->id);
1581
1582 if (evutil_make_socket_nonblocking(fd) < 0) {
1583 LOG(&null_worker, "Failed to set client socket to non-blocking : %m");
1584 close(fd);
1585 return;
1586 }
1587
1588 if ((client = calloc(1, sizeof(*client))) == NULL) {
1589 LOG(&null_worker, "Failed to allocate memory for client state : %m");
1590 close(fd);
1591 return;
1592 }
1593
1594 client->id = clicount;
1595 client->fd = fd;
1596 client->worker = worker;
1597 client->state = SPOA_ST_CONNECTING;
1598 client->status_code = SPOE_FRM_ERR_NONE;
1599 client->max_frame_size = max_frame_size;
1600 client->engine = NULL;
1601 client->pipelining = false;
1602 client->async = false;
1603 client->incoming_frame = NULL;
1604 client->outgoing_frame = NULL;
1605 LIST_INIT(&client->processing_frames);
1606 LIST_INIT(&client->outgoing_frames);
1607
1608 LIST_ADDQ(&worker->clients, &client->by_worker);
1609
1610 worker->nbclients++;
1611
1612 if (event_assign(&client->read_frame_event, worker->base, fd,
1613 EV_READ|EV_PERSIST, read_frame_cb, client) < 0 ||
1614 event_assign(&client->write_frame_event, worker->base, fd,
1615 EV_WRITE|EV_PERSIST, write_frame_cb, client) < 0) {
1616 LOG(&null_worker, "Failed to create client events");
1617 release_client(client);
1618 return;
1619 }
1620 event_add(&client->read_frame_event, NULL);
1621 }
1622
1623 static void *
worker_function(void * data)1624 worker_function(void *data)
1625 {
1626 struct client *client, *cback;
1627 struct spoe_frame *frame, *fback;
1628 struct worker *worker = data;
1629
1630 DEBUG(worker, "Worker ready to process client messages");
1631 event_base_dispatch(worker->base);
1632
1633 list_for_each_entry_safe(client, cback, &worker->clients, by_worker) {
1634 release_client(client);
1635 }
1636
1637 list_for_each_entry_safe(frame, fback, &worker->frames, list) {
1638 LIST_DEL(&frame->list);
1639 free(frame);
1640 }
1641
1642 event_free(worker->monitor_event);
1643 event_base_free(worker->base);
1644 DEBUG(worker, "Worker is stopped");
1645 pthread_exit(&null_worker);
1646 }
1647
1648
1649 static int
parse_processing_delay(const char * str)1650 parse_processing_delay(const char *str)
1651 {
1652 unsigned long value;
1653
1654 value = 0;
1655 while (1) {
1656 unsigned int j;
1657
1658 j = *str - '0';
1659 if (j > 9)
1660 break;
1661 str++;
1662 value *= 10;
1663 value += j;
1664 }
1665
1666 switch (*str) {
1667 case '\0': /* no unit = millisecond */
1668 value *= 1000;
1669 break;
1670 case 's': /* second */
1671 value *= 1000000;
1672 str++;
1673 break;
1674 case 'm': /* millisecond : "ms" */
1675 if (str[1] != 's')
1676 return -1;
1677 value *= 1000;
1678 str += 2;
1679 break;
1680 case 'u': /* microsecond : "us" */
1681 if (str[1] != 's')
1682 return -1;
1683 str += 2;
1684 break;
1685 default:
1686 return -1;
1687 }
1688 if (*str)
1689 return -1;
1690
1691 processing_delay.tv_sec = (time_t)(value / 1000000);
1692 processing_delay.tv_usec = (suseconds_t)(value % 1000000);
1693 return 0;
1694 }
1695
1696
1697 static void
usage(char * prog)1698 usage(char *prog)
1699 {
1700 fprintf(stderr,
1701 "Usage : %s [OPTION]...\n"
1702 " -h Print this message\n"
1703 " -f <config-file> Mod Defender configuration file\n"
1704 " -l <log-file> Mod Defender log file\n"
1705 " -d Enable the debug mode\n"
1706 " -m <max-frame-size> Specify the maximum frame size (default : %u)\n"
1707 " -p <port> Specify the port to listen on (default : %d)\n"
1708 " -n <num-workers> Specify the number of workers (default : %d)\n"
1709 " -c <capability> Enable the support of the specified capability\n"
1710 " -t <time> Set a delay to process a message (default: 0)\n"
1711 " The value is specified in milliseconds by default,\n"
1712 " but can be in any other unit if the number is suffixed\n"
1713 " by a unit (us, ms, s)\n"
1714 "\n"
1715 " Supported capabilities: fragmentation, pipelining, async\n",
1716 prog, MAX_FRAME_SIZE, DEFAULT_PORT, NUM_WORKERS);
1717 }
1718
1719 int
main(int argc,char ** argv)1720 main(int argc, char **argv)
1721 {
1722 struct event_base *base = NULL;
1723 struct event *signal_event = NULL, *accept_event = NULL;
1724 int opt, i, fd = -1;
1725 const char *config_file = NULL;
1726 const char *log_file = NULL;
1727
1728 // TODO: add '-t <processing-time>' option
1729 while ((opt = getopt(argc, argv, "hf:l:dm:n:p:c:t:")) != -1) {
1730 switch (opt) {
1731 case 'h':
1732 usage(argv[0]);
1733 return EXIT_SUCCESS;
1734 case 'f':
1735 config_file = optarg;
1736 break;
1737 case 'l':
1738 log_file = optarg;
1739 break;
1740 case 'd':
1741 debug = true;
1742 break;
1743 case 'm':
1744 max_frame_size = atoi(optarg);
1745 break;
1746 case 'n':
1747 num_workers = atoi(optarg);
1748 break;
1749 case 'p':
1750 server_port = atoi(optarg);
1751 break;
1752 case 'c':
1753 if (!strcmp(optarg, "pipelining"))
1754 pipelining = true;
1755 else if (!strcmp(optarg, "async"))
1756 async = true;
1757 else if (!strcmp(optarg, "fragmentation"))
1758 fragmentation = true;
1759 else
1760 fprintf(stderr, "WARNING: unsupported capability '%s'\n", optarg);
1761 break;
1762 case 't':
1763 if (!parse_processing_delay(optarg))
1764 break;
1765 fprintf(stderr, "%s: failed to parse time '%s'.\n", argv[0], optarg);
1766 fprintf(stderr, "Try '%s -h' for more information.\n", argv[0]);
1767 return EXIT_FAILURE;
1768 default:
1769 usage(argv[0]);
1770 return EXIT_FAILURE;
1771 }
1772 }
1773
1774 if (!defender_init(config_file, log_file))
1775 goto error;
1776
1777 if (num_workers <= 0) {
1778 LOG(&null_worker, "%s : Invalid number of workers '%d'\n",
1779 argv[0], num_workers);
1780 goto error;
1781 }
1782
1783 if (server_port <= 0) {
1784 LOG(&null_worker, "%s : Invalid port '%d'\n",
1785 argv[0], server_port);
1786 goto error;
1787 }
1788
1789 if (evthread_use_pthreads() < 0) {
1790 LOG(&null_worker, "No pthreads support for libevent");
1791 goto error;
1792 }
1793
1794 if ((base = event_base_new()) == NULL) {
1795 LOG(&null_worker, "Failed to initialize libevent : %m");
1796 goto error;
1797 }
1798
1799 signal(SIGPIPE, SIG_IGN);
1800
1801 if ((fd = create_server_socket()) < 0) {
1802 LOG(&null_worker, "Failed to create server socket");
1803 goto error;
1804 }
1805 if (evutil_make_socket_nonblocking(fd) < 0) {
1806 LOG(&null_worker, "Failed to set server socket to non-blocking");
1807 goto error;
1808 }
1809
1810 if ((workers = calloc(num_workers, sizeof(*workers))) == NULL) {
1811 LOG(&null_worker, "Failed to set allocate memory for workers");
1812 goto error;
1813 }
1814
1815 for (i = 0; i < num_workers; ++i) {
1816 struct worker *w = &workers[i];
1817
1818 w->id = i+1;
1819 w->nbclients = 0;
1820 LIST_INIT(&w->engines);
1821 LIST_INIT(&w->clients);
1822 LIST_INIT(&w->frames);
1823
1824 if ((w->base = event_base_new()) == NULL) {
1825 LOG(&null_worker,
1826 "Failed to initialize libevent for worker %02d : %m",
1827 w->id);
1828 goto error;
1829 }
1830
1831 w->monitor_event = event_new(w->base, fd, EV_PERSIST,
1832 worker_monitor_cb, (void *)w);
1833 if (w->monitor_event == NULL ||
1834 event_add(w->monitor_event, (struct timeval[]){{5,0}}) < 0) {
1835 LOG(&null_worker,
1836 "Failed to create monitor event for worker %02d",
1837 w->id);
1838 goto error;
1839 }
1840
1841 if (pthread_create(&w->thread, NULL, worker_function, (void *)w)) {
1842 LOG(&null_worker,
1843 "Failed to start thread for worker %02d : %m",
1844 w->id);
1845 }
1846 DEBUG(&null_worker, "Worker %02d initialized", w->id);
1847 }
1848
1849 accept_event = event_new(base, fd, EV_READ|EV_PERSIST, accept_cb,
1850 (void *)base);
1851 if (accept_event == NULL || event_add(accept_event, NULL) < 0) {
1852 LOG(&null_worker, "Failed to create accept event : %m");
1853 }
1854
1855 signal_event = evsignal_new(base, SIGINT, signal_cb, (void *)base);
1856 if (signal_event == NULL || event_add(signal_event, NULL) < 0) {
1857 LOG(&null_worker, "Failed to create signal event : %m");
1858 }
1859
1860 DEBUG(&null_worker,
1861 "Server is ready"
1862 " [fragmentation=%s - pipelining=%s - async=%s - debug=%s - max-frame-size=%u]",
1863 (fragmentation?"true":"false"), (pipelining?"true":"false"), (async?"true":"false"),
1864 (debug?"true":"false"), max_frame_size);
1865 event_base_dispatch(base);
1866
1867 for (i = 0; i < num_workers; i++) {
1868 struct worker *w = &workers[i];
1869
1870 pthread_join(w->thread, NULL);
1871 DEBUG(&null_worker, "Worker %02d terminated", w->id);
1872 }
1873
1874 free(workers);
1875 event_free(signal_event);
1876 event_free(accept_event);
1877 event_base_free(base);
1878 close(fd);
1879 return EXIT_SUCCESS;
1880
1881 error:
1882 if (workers != NULL)
1883 free(workers);
1884 if (signal_event != NULL)
1885 event_free(signal_event);
1886 if (accept_event != NULL)
1887 event_free(accept_event);
1888 if (base != NULL)
1889 event_base_free(base);
1890 if (fd != -1)
1891 close(fd);
1892 return EXIT_FAILURE;
1893 }
1894