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