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